Stop building Death in Space into the platform; make backtrack undo the turn's writes (RS-1, RS-2, P-7)

docs/architecture/GM_AS_AUTHORITY.md

@@ -196,7 +196,6 @@ involved.
 - `packages/agents/tests/test_narrator_refines.py` (refinement path)
 - `packages/agents/tests/test_resolver_gm_loop.py` (GM-driven resolver)
 - `scripts/e2e_full_loop.py` (the E2E harness)
-- `scripts/death_in_space_game_system.py` (DiS fixture)
 - `docs/testing/E2E_GM_AUTHORITY_<date>.md` (per-session analysis)
 
 ## Post-refactor call: do we still need a small custom SLM?

packages/agents/src/monitor_agents/loops/scene_loop.py

@@ -1840,6 +1840,13 @@ async def backtrack(self) -> dict[str, Any]:
                 # Undoing Turn 1 restores the pre-turn-1 initial scene state.
                 restored_values = {**dict(history[-1].values or {}), **base_defaults}
 
+            # The turn also wrote outside the checkpoint. Undo that first, and only
+            # rewind the checkpoint once it worked, so the two never disagree.
+            restore_point = prev_turn_state if prev_turn_state is not None else history[-1]
+            rewind = await _rewind_scene_writes(self.scene_id, restore_point.created_at, restored_values)
+            if "error" in rewind:
+                return rewind
+
             restored_suggestions: list[ImageSuggestion] = []
             for item in restored_values.get("image_suggestions") or []:
                 if isinstance(item, ImageSuggestion):
@@ -1853,8 +1860,55 @@ async def backtrack(self) -> dict[str, Any]:
                 config,  # type: ignore[arg-type]
                 restored_values,
             )
+            PersistenceService.clear_scene_runtime_cache(self.scene_id)
+
+            return {**restored_values, "undone": rewind}
+
+
+async def _rewind_scene_writes(
+    scene_id: UUID, restore_point: str | None, restored_values: dict[str, Any]
+) -> dict[str, Any]:
+    """Remove what the scene wrote after the restore point; P-7.
 
-            return restored_values
+    Returns what was undone, or ``{"error": ...}``. A refusal changes nothing. A database
+    failure can leave the rewind part done, and running it again finishes it, since it
+    deletes by the same boundary.
+    """
+    import anyio
+    from monitor_data.schemas.vectors import VectorDeleteRequest
+    from monitor_data.tools.mongodb_tools.scene_rewind import SceneRewindRefused, mongodb_rewind_scene
+    from monitor_data.tools.qdrant_tools import qdrant_delete
+
+    if not restore_point:
+        return {"error": "Cannot backtrack: the checkpoint has no time, so its later writes can't be found."}
+    since = datetime.fromisoformat(restore_point)
+    earlier: dict[str, dict[str, Any]] = {}
+    snapshot = restored_values.get("working_state")
+    if isinstance(snapshot, dict) and snapshot.get("state_id"):
+        earlier[str(snapshot["state_id"])] = snapshot
+    try:
+        result = await anyio.to_thread.run_sync(
+            functools.partial(mongodb_rewind_scene, scene_id, since, working_states=earlier)
+        )
+    except SceneRewindRefused as exc:
+        return {"error": str(exc)}
+    except Exception as exc:  # noqa: BLE001  # reason: reported to the player; the checkpoint stays put and a retry finishes the job
+        logger.warning("backtrack: rewinding scene %s failed: %s", scene_id, exc)
+        return {
+            "error": (
+                "The turn may be partly undone: the database stopped responding. "
+                "Run /backtrack again to finish; it picks up where this stopped."
+            )
+        }
+    for memory_id in result.memory_ids_removed:
+        try:
+            await anyio.to_thread.run_sync(
+                qdrant_delete, VectorDeleteRequest(collection="memories", id=UUID(memory_id))
+            )
+        except Exception as exc:  # noqa: BLE001  # reason: the memory is gone from MongoDB; a stale vector is reported, not fatal
+            logger.warning("backtrack: memory vector %s not deleted: %s", memory_id, exc)
+            result.not_undone.append(f"memory {memory_id}: its search vector is still stored")
+    return result.model_dump()
 
 
 _NON_COMBATANT_ENTITY_TYPES = frozenset(

packages/agents/tests/test_scene_state_carryover.py

@@ -99,8 +99,30 @@ async def _fake_read(fn: Any, *_args: Any, **_kwargs: Any) -> Any:
     return []
 
 
+class _Rewinds:
+    """Stands in for the data-layer rewind, which has its own tests, and records each call."""
+
+    def __init__(self) -> None:
+        self.calls: list[tuple[Any, Any, dict[str, Any]]] = []
+        self.refuse: str | None = None
+        self.memory_ids: list[str] = []
+
+    def __call__(self, scene_id: Any, since: Any, *, working_states: Any = None) -> Any:
+        from monitor_data.tools.mongodb_tools.scene_rewind import SceneRewindRefused, SceneRewindResult
+
+        if self.refuse:
+            raise SceneRewindRefused(self.refuse)
+        self.calls.append((scene_id, since, working_states or {}))
+        return SceneRewindResult(memory_ids_removed=list(self.memory_ids))
+
+
 @pytest.fixture
-def world():
+def rewinds():
+    return _Rewinds()
+
+
+@pytest.fixture
+def world(rewinds):
     saver = InMemorySaver()
     state = _World()
 
@@ -117,7 +139,10 @@ def conn(_uri: str):
         patch.object(sl.PersistenceService, "persist_working_state", AsyncMock(return_value={})),
         patch("monitor_agents.foreshadowing.agent.ForeshadowingAgent.propose", AsyncMock(return_value={})),
         patch.object(sl, "seed_npc_profiles_for_accepted_proposals", AsyncMock(return_value=0)),
+        patch("monitor_data.tools.mongodb_tools.scene_rewind.mongodb_rewind_scene", rewinds),
+        patch("monitor_data.tools.qdrant_tools.qdrant_delete") as qdrant_delete,
     ):
+        state.qdrant_delete = qdrant_delete  # type: ignore[attr-defined]
         yield state
 
 
@@ -457,6 +482,73 @@ async def test_backtrack_reverts_full_turn_and_steps_back_repeatedly(world: _Wor
     assert "error" in await loop.backtrack()
 
 
+async def _checkpoints(loop: sl.SceneLoop) -> list[Any]:
+    with sl.MongoDBSaver.from_conn_string("") as saver:
+        compiled = loop._graph.compile(checkpointer=saver)
+        config = {"configurable": {"thread_id": str(loop.scene_id)}}
+        return [snap async for snap in compiled.aget_state_history(config)]  # type: ignore[arg-type]
+
+
+@pytest.mark.asyncio
+async def test_backtrack_undoes_the_turns_writes_since_the_previous_turn(world: _World, rewinds: _Rewinds) -> None:
+    from datetime import datetime
+
+    loop = _loop()
+    await loop.run("turn one")
+    turn_one_end = next(s for s in await _checkpoints(loop) if not s.next)
+    await loop.run("turn two")
+    rewinds.memory_ids = [str(uuid4())]
+
+    restored = await loop.backtrack()
+
+    ((scene_id, since, _),) = rewinds.calls
+    assert scene_id == loop.scene_id
+    assert since == datetime.fromisoformat(turn_one_end.created_at)
+    assert restored["turns_count"] == 1 and restored["undone"]["memory_ids_removed"] == rewinds.memory_ids
+    (request,) = [c.args[0] for c in world.qdrant_delete.call_args_list]  # type: ignore[attr-defined]
+    assert request.collection == "memories" and str(request.id) == rewinds.memory_ids[0]
+
+
+@pytest.mark.asyncio
+async def test_a_refused_rewind_leaves_the_turn_in_place(world: _World, rewinds: _Rewinds) -> None:
+    loop = _loop()
+    await loop.run("turn one")
+    await loop.run("turn two")
+    rewinds.refuse = "1 change(s) from this turn were already accepted into canon"
+
+    assert (await loop.backtrack()) == {"error": "1 change(s) from this turn were already accepted into canon"}
+    rewinds.refuse = None
+    # The checkpoint didn't move, so the next backtrack still undoes turn two.
+    assert (await loop.backtrack())["turns_count"] == 1
+
+
+@pytest.mark.asyncio
+async def test_a_database_failure_says_to_run_backtrack_again(world: _World, rewinds: _Rewinds) -> None:
+    loop = _loop()
+    await loop.run("turn one")
+    await loop.run("turn two")
+    with patch("monitor_data.tools.mongodb_tools.scene_rewind.mongodb_rewind_scene", side_effect=OSError("down")):
+        failed = await loop.backtrack()
+    assert "partly undone" in failed["error"] and "/backtrack again" in failed["error"]
+    assert (await loop.backtrack())["turns_count"] == 1
+
+
+@pytest.mark.asyncio
+async def test_backtrack_passes_the_earlier_working_state_to_restore(world: _World, rewinds: _Rewinds) -> None:
+    earlier = {"state_id": "ws-1", "current_stats": {"xp": 5}, "resources": {"HP": 9}}
+    sl.PersistenceService.persist_working_state.side_effect = [  # type: ignore[attr-defined]
+        {"working_state": earlier},
+        {"working_state": {**earlier, "resources": {"HP": 3}}},
+    ]
+    world.resolutions = [{"resolution_type": "dice"}, {"resolution_type": "dice"}]
+    loop = _loop()
+    await loop.run("turn one")
+    await loop.run("turn two")
+    await loop.backtrack()
+    ((_, _, states),) = rewinds.calls
+    assert states == {"ws-1": earlier}
+
+
 @pytest.mark.asyncio
 async def test_multi_turn_combat_carries_round_and_hp_and_persists_deltas(world: _World) -> None:
     from monitor_agents.loops import combat_loop as cl

packages/cli/src/monitor_cli/commands/play.py

@@ -238,6 +238,8 @@ async def _run_play(
                 else:
                     turn_number = max(0, turn_number - 1)
                     console.print("[green]State restored. You can re-enter your action.[/green]")
+                    for item in (backtrack_result.get("undone") or {}).get("not_undone") or []:
+                        console.print(f"[yellow]Not undone:[/yellow] {item}")
                 continue
 
             if user_input.lower().startswith("/flashback "):