diff --git a/api/core/workflow_copilot/ports.py b/api/core/workflow_copilot/ports.py
index fad6bdd5501..54ef35214f1 100644
--- a/api/core/workflow_copilot/ports.py
+++ b/api/core/workflow_copilot/ports.py
@@ -124,6 +124,8 @@ class DifyPort(Protocol):
 
     def publish(self, app_id: str, actor: Actor) -> None: ...
 
+    def restore_graph(self, app_id: str, actor: Actor, graph: Graph) -> str: ...
+
 
 @runtime_checkable
 class Repository(Protocol):
diff --git a/api/services/workflow_copilot/dify_port.py b/api/services/workflow_copilot/dify_port.py
index e7e6e3268e3..98b7f9bd081 100644
--- a/api/services/workflow_copilot/dify_port.py
+++ b/api/services/workflow_copilot/dify_port.py
@@ -224,6 +224,34 @@ class WorkflowServiceDifyPort:
                 changed_nodes=changed_nodes, new_hash=updated.unique_hash, changes=changes, scope=scope
             )
 
+    def restore_graph(self, app_id: str, actor: Actor, graph: Graph) -> str:
+        """Restore the draft to a prior snapshot graph (Slice 4 revert): write
+        the whole graph back via the same graph-only sync apply_repair uses.
+        Returns the new draft hash. Raises HashMismatchError if the draft
+        changed since the caller read it (same contract as apply_repair)."""
+        with _session_factory()() as session:
+            account = resolve_account(session, actor)
+            app = load_app(session, app_id, actor)
+            workflow = _load_draft_workflow_or_raise(app, session=session)
+            try:
+                updated = WorkflowService().sync_draft_workflow(
+                    app_model=app,
+                    graph=graph,
+                    features=dict(workflow.features_dict),
+                    unique_hash=workflow.unique_hash,
+                    account=account,
+                    environment_variables=[],
+                    conversation_variables=workflow.conversation_variables,
+                    session=session,
+                    graph_only=True,
+                    sync_agent_bindings=False,
+                    commit=True,
+                    preserve_environment_variables=True,
+                )
+            except WorkflowHashNotEqualError as exc:
+                raise HashMismatchError(f"draft workflow changed since read: {app_id}") from exc
+            return updated.unique_hash
+
     def run_draft(self, app_id: str, actor: Actor, inputs: Inputs, on_event: Callable[[NodeEvent], None]) -> Run:
         with _session_factory()() as session:
             account = resolve_account(session, actor)
diff --git a/api/tests/unit_tests/core/workflow_copilot/fakes.py b/api/tests/unit_tests/core/workflow_copilot/fakes.py
index 9c757bbf6bd..cc04affde03 100644
--- a/api/tests/unit_tests/core/workflow_copilot/fakes.py
+++ b/api/tests/unit_tests/core/workflow_copilot/fakes.py
@@ -287,6 +287,11 @@ class FakeDifyPort:
     def publish(self, _app_id: str, _actor: Actor) -> None:
         self.published = True
 
+    def restore_graph(self, _app_id: str, _actor: Actor, graph: Graph) -> str:
+        self.graph = copy.deepcopy(graph)
+        self.hash = "h-restored"
+        return self.hash
+
 
 # ---- stub CopilotAgent ----------------------------------------------------
 
@@ -439,6 +444,11 @@ class FakeBuildDifyPort:
     def publish(self, _app_id: str, _actor: Actor) -> None:
         self.published = True
 
+    def restore_graph(self, _app_id: str, _actor: Actor, graph: Graph) -> str:
+        self.graph = copy.deepcopy(graph)
+        self.hash = "h-restored"
+        return self.hash
+
 
 # ---- fake DifyPort pre-seeded with an existing graph (for Edit) -----------
 
diff --git a/api/tests/unit_tests/core/workflow_copilot/test_lifecycle.py b/api/tests/unit_tests/core/workflow_copilot/test_lifecycle.py
new file mode 100644
index 00000000000..97fc24e83f7
--- /dev/null
+++ b/api/tests/unit_tests/core/workflow_copilot/test_lifecycle.py
@@ -0,0 +1,38 @@
+"""Slice 4 session-lifecycle-core tests: restore_graph, invalidation, revert,
+checkpoint surfacing, pause/resume."""
+
+from core.workflow_copilot.models import Actor
+from core.workflow_copilot.ports import DifyPort
+from tests.unit_tests.core.workflow_copilot.fakes import FakeBuildDifyPort, FakeDifyPort
+
+
+def _actor() -> Actor:
+    return Actor(account_id="a", tenant_id="t")
+
+
+def test_fake_dify_port_restore_graph_swaps_graph_and_returns_hash():
+    dify = FakeDifyPort()
+    dify.graph = {"nodes": [{"id": "x"}]}
+    snapshot = {"nodes": [{"id": "start"}, {"id": "end"}]}
+    new_hash = dify.restore_graph("app", _actor(), snapshot)
+    assert isinstance(new_hash, str)
+    assert new_hash
+    got, got_hash = dify.read_graph("app", _actor())
+    assert got == snapshot
+    assert got_hash == new_hash
+    # deep-copied: mutating the caller's graph does not corrupt the fake
+    snapshot["nodes"].append({"id": "leak"})
+    assert len(dify.read_graph("app", _actor())[0]["nodes"]) == 2
+
+
+def test_fake_build_dify_port_restore_graph_swaps_graph():
+    dify = FakeBuildDifyPort()
+    dify.graph = {"nodes": [{"id": "a"}], "edges": []}
+    snapshot = {"nodes": [], "edges": []}
+    dify.restore_graph("app", _actor(), snapshot)
+    assert dify.read_graph("app", _actor())[0] == {"nodes": [], "edges": []}
+
+
+def test_fake_dify_ports_still_satisfy_dify_port_protocol():
+    assert isinstance(FakeDifyPort(), DifyPort)
+    assert isinstance(FakeBuildDifyPort(), DifyPort)
diff --git a/api/tests/unit_tests/core/workflow_copilot/test_ports.py b/api/tests/unit_tests/core/workflow_copilot/test_ports.py
index 8db74edcd7a..e321efd5f9b 100644
--- a/api/tests/unit_tests/core/workflow_copilot/test_ports.py
+++ b/api/tests/unit_tests/core/workflow_copilot/test_ports.py
@@ -159,6 +159,9 @@ class _StubDifyPort:
     def publish(self, _app_id, _actor):
         return None
 
+    def restore_graph(self, _app_id, _actor, _graph):
+        return "hash-restored"
+
 
 class _StubRepository:
     """A trivial conforming Repository."""