Journal promotions for crash recovery; surface rejected saves and sync playground tabs

CHANGELOG.md

@@ -8,10 +8,16 @@ releases could contain breaking changes when called out explicitly.
 
 ### Added
 
+- Keep multiple playground tabs on the same workspace in sync: a tab without unsaved edits adopts a newer revision saved by another tab, and a tab with unsaved edits is marked memory-only instead of silently losing the race.
+
 ### Changed
 
 ### Fixed
 
+- Recover from a compilation that was killed mid-promotion: the next compilation rolls the workspace back to its pre-promotion files using a journal kept under `.modelable/locks`. An unreadable journal blocks compilation instead of being ignored.
+
+- Show the playground as memory-only instead of saved when a workspace save is rejected because a newer revision is already stored (for example from another tab).
+
 - Keep conversation compilation previews recoverable after lock-contention and filesystem errors, returning an error reply and preserving pending actions instead of failing the turn.
 
 - Preserve affected artifact references such as `go.mod` in conversation previews without treating them as source-model locations.

cli/src/modelable/operations/compilation.py

@@ -307,6 +307,9 @@ def _execute_direct_staged(
         workspace_root: Path,
         staging_dir: Path,
     ) -> DirectCompilationResult:
+        # Finish or undo a promotion that a dead process left behind before
+        # reading any workspace state to stage from.
+        self.transaction_factory(workspace_root).recover()
         staged_output = staging_dir / "output"
         staged_registry_ids = staging_dir / "registry-ids.lock"
         staged_enum_numbers = staging_dir / "enum-numbers.lock"

cli/src/modelable/operations/file_transaction.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import hashlib
+import json
 import os
 import shutil
 import sys
@@ -57,6 +58,10 @@ def __init__(
         super().__init__(message)
 
 
+class _JournalError(Exception):
+    """The promotion journal was unreadable or could not be fully replayed."""
+
+
 class FileTransactionCommittedError(Exception):
     """Promotion committed, but post-commit cleanup was incomplete."""
 
@@ -132,6 +137,13 @@ def promote(
             release_errors = self._release_lock()
             raise FileTransactionError(error, release_errors) from error
 
+        try:
+            self._recover_interrupted_promotion()
+        except Exception as error:
+            recovery_errors = [RollbackError(self._journal_path(), error)] if isinstance(error, _JournalError) else []
+            recovery_errors.extend(self._release_lock())
+            raise FileTransactionError(error, recovery_errors) from error
+
         if validate is not None:
             try:
                 validate()
@@ -176,6 +188,8 @@ def promote(
             if validate is not None:
                 validate()
 
+            self._write_journal(promotions, created_directories)
+
             for promotion in promotions:
                 metadata_source = (
                     promotion.file.destination if promotion.backup is not None else promotion.file.staged_path
@@ -187,12 +201,19 @@ def promote(
             for promotion in promotions:
                 if _file_hash(promotion.file.destination) != promotion.file.content_hash:
                     raise OSError(f"post-write hash verification failed for {promotion.file.destination}")
+
+            # Removing the journal is the commit point: until it is gone, a
+            # crash is rolled back by the next transaction.
+            self._unlink(self._journal_path(), missing_ok=True)
         except Exception as error:
             rollback_errors = self._rollback(
                 promoted,
                 promotions,
                 created_directories,
             )
+            if not rollback_errors:
+                with suppress(OSError):
+                    self._unlink(self._journal_path(), missing_ok=True)
             rollback_errors.extend(self._release_lock())
             raise FileTransactionError(error, rollback_errors) from error
 
@@ -203,6 +224,111 @@ def promote(
             raise FileTransactionCommittedError(written_paths, cleanup_errors)
         return written_paths
 
+    def recover(self) -> None:
+        """Roll back a promotion interrupted by a dead process, if one left a journal.
+
+        Callers that read workspace state before promoting call this first, so
+        they never stage output derived from a half-promoted workspace.
+        """
+        if self.workspace_root is None:
+            raise ValueError("recover() requires workspace_root")
+        journal = self.workspace_root / ".modelable" / "locks" / "compilation.journal"
+        if not journal.exists():
+            return
+        self._acquire_lock(self.workspace_root)
+        try:
+            self._recover_interrupted_promotion()
+        except Exception as error:
+            recovery_errors = [RollbackError(journal, error)] if isinstance(error, _JournalError) else []
+            recovery_errors.extend(self._release_lock())
+            raise FileTransactionError(error, recovery_errors) from error
+        release_errors = self._release_lock()
+        if release_errors:
+            raise FileTransactionError(OSError("could not release the workspace lock"), release_errors)
+
+    def _journal_path(self) -> Path:
+        assert self._lock_path is not None
+        return self._lock_path.with_name("compilation.journal")
+
+    def _write_journal(
+        self,
+        promotions: Sequence[_Promotion],
+        created_directories: Sequence[Path],
+    ) -> None:
+        journal = {
+            "version": 1,
+            "entries": [
+                {
+                    "destination": str(item.file.destination),
+                    "temporary": str(item.temporary),
+                    "backup": str(item.backup) if item.backup is not None else None,
+                    "hash": item.file.content_hash,
+                }
+                for item in promotions
+            ],
+            "created_directories": [str(path) for path in dict.fromkeys(created_directories)],
+        }
+        path = self._journal_path()
+        temporary = path.with_name(f"{path.name}.{self._token}.tmp")
+        descriptor = os.open(
+            temporary,
+            os.O_CREAT | os.O_EXCL | os.O_WRONLY | getattr(os, "O_BINARY", 0),
+            0o600,
+        )
+        try:
+            os.write(descriptor, json.dumps(journal).encode())
+            os.fsync(descriptor)
+        finally:
+            os.close(descriptor)
+        os.replace(temporary, path)
+
+    def _recover_interrupted_promotion(self) -> None:
+        """Roll back a promotion whose process died before it committed.
+
+        Runs under the workspace lock. A journal that still exists means the
+        commit point was never reached, so every destination is returned to its
+        pre-promotion state. A journal that cannot be read blocks the
+        transaction and is kept for inspection.
+        """
+        path = self._journal_path()
+        try:
+            data = path.read_bytes()
+        except FileNotFoundError:
+            return
+        try:
+            journal = json.loads(data)
+            entries = [
+                (
+                    Path(entry["destination"]),
+                    Path(entry["temporary"]),
+                    Path(entry["backup"]) if entry["backup"] is not None else None,
+                    str(entry["hash"]),
+                )
+                for entry in journal["entries"]
+            ]
+            created = [Path(item) for item in journal["created_directories"]]
+        except (ValueError, KeyError, TypeError) as error:
+            raise _JournalError(f"Unreadable compilation journal {path}: {error}") from error
+
+        errors: list[RollbackError] = []
+        for destination, temporary, backup, content_hash in reversed(entries):
+            try:
+                if backup is not None:
+                    if backup.exists():
+                        self._replace(backup, destination)
+                elif destination.exists() and _file_hash(destination) == content_hash:
+                    self._unlink(destination, missing_ok=True)
+                self._unlink(temporary, missing_ok=True)
+            except Exception as error:
+                errors.append(RollbackError(destination, error))
+        for directory in reversed(created):
+            with suppress(OSError):
+                directory.rmdir()
+        if errors:
+            details = os.linesep.join(f"  - {item}" for item in errors)
+            raise _JournalError(f"Could not recover interrupted compilation promotion:{os.linesep}{details}")
+        self._unlink(path, missing_ok=True)
+
     def _acquire_lock(self, root: Path) -> None:
         lock_path = root / ".modelable" / "locks" / "compilation.lock"
         self._lock_directories = (lock_path.parent, lock_path.parent.parent)

cli/tests/test_compilation_service.py

@@ -1717,3 +1717,72 @@ def fail_staging_cleanup(path: Path) -> None:
     if pending is not None:
         audit = tmp_path / ".modelable" / "audit" / "compilations" / f"{pending.action_id}.json"
         assert audit.exists()
+
+
+class _SimulatedProcessDeath(BaseException):
+    """Skips every ``except Exception`` handler, like a killed process."""
+
+
+def test_direct_compilation_recovers_an_interrupted_promotion_before_staging(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    events: list[str] = []
+
+    class RecordingTransaction(FileTransaction):
+        def recover(self) -> None:
+            events.append("recover")
+            super().recover()
+
+        def promote(self, files, **kwargs):  # type: ignore[no-untyped-def]
+            events.append("promote")
+            return super().promote(files, **kwargs)
+
+    source = write_workspace(tmp_path)
+    request = request_for(tmp_path, source, "rust")
+    CompilationService(
+        transaction_factory=lambda root: RecordingTransaction(workspace_root=root),
+    ).execute_direct(request)
+
+    assert events == ["recover", "promote"]
+
+
+def test_direct_compilation_after_a_crashed_promotion_matches_a_clean_run(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    def crash_on_second_replace(source_path, destination_path):
+        if ".modelable-tmp-" in Path(source_path).name:
+            if crash_on_second_replace.count == 1:  # type: ignore[attr-defined]
+                raise _SimulatedProcessDeath
+            crash_on_second_replace.count += 1  # type: ignore[attr-defined]
+        os.replace(source_path, destination_path)
+
+    crash_on_second_replace.count = 0  # type: ignore[attr-defined]
+
+    control = tmp_path / "control"
+    control.mkdir()
+    csrc = write_workspace(control)
+    CompilationService().execute_direct(request_for(control, csrc, "rust"))
+
+    crashed = tmp_path / "crashed"
+    crashed.mkdir()
+    source = write_workspace(crashed)
+    with pytest.raises(_SimulatedProcessDeath):
+        CompilationService(
+            transaction_factory=lambda root: FileTransaction(workspace_root=root, replace=crash_on_second_replace),
+        ).execute_direct(request_for(crashed, source, "rust"))
+    monkeypatch.setattr("modelable.operations.file_transaction._process_alive", lambda pid: False)
+    monkeypatch.setattr("modelable.registry.ids._process_alive", lambda pid: False, raising=False)
+
+    CompilationService().execute_direct(request_for(crashed, source, "rust"))
+
+    def snapshot(root: Path) -> dict[str, bytes]:
+        return {
+            path.relative_to(root).as_posix(): path.read_bytes()
+            for path in sorted(root.rglob("*"))
+            if path.is_file() and ".modelable-" not in path.name and path.name != "registry.db"
+        }
+
+    assert snapshot(crashed) == snapshot(control)
+    assert not list(crashed.rglob("compilation.journal"))