fix: address fork() bugs found in code review

Fix four issues identified by @VascoSch92:

1. Path-doubling: fork() was computing fork_persistence as
   base/FORK_HEX, but __init__→get_persistence_dir() appends the hex
   again, producing base/FORK_HEX/FORK_HEX. Fix: pass only the base
   directory and let __init__ append the hex.

2. Missing FIFOLock: fork() reads mutable state (events, stats,
   agent_state, activated_knowledge_skills) without acquiring the
   state lock, risking torn reads during concurrent run(). Fix: wrap
   all state reads in with self._state:.

3. Orphaned persistence dir: if _start_event_service raises after
   fork persistence is written to disk, the directory is never cleaned
   up. Fix: add try/except with safe_rmtree rollback.

4. No duplicate-ID check: client-supplied fork ID could clobber an
   active conversation. Fix: reject duplicate IDs with ValueError,
   surface as HTTP 409 Conflict at the router layer.

Add regression tests:
- test_fork_persistence_path_no_doubling: verifies fork dir is a
  sibling of source, not nested
- test_fork_persisted_events_survive_reload: end-to-end persistence
  round-trip (close fork, reopen from disk, verify events)
- test_fork_conversation_duplicate_id_returns_409: router returns 409
  for duplicate fork IDs

Co-authored-by: openhands <openhands@all-hands.dev>

Author: openhands-agent

openhands-agent-server/openhands/agent_server/conversation_router.py

@@ -400,6 +400,7 @@ async def condense_conversation(
     responses={
         201: {"description": "Forked conversation created"},
         404: {"description": "Source conversation not found"},
+        409: {"description": "Fork ID already in use"},
     },
     status_code=status.HTTP_201_CREATED,
 )
@@ -414,15 +415,21 @@ async def fork_conversation(
     Calling ``run`` on the fork resumes from the copied state, meaning
     the agent has full event memory of the source conversation.
     """
-    info = await conversation_service.fork_conversation(
-        conversation_id,
-        fork_id=request.id,
-        title=request.title,
-        tags=request.tags if request.tags is not None else None,
-        reset_metrics=request.reset_metrics,
-    )
+    try:
+        info = await conversation_service.fork_conversation(
+            conversation_id,
+            fork_id=request.id,
+            title=request.title,
+            tags=request.tags if request.tags is not None else None,
+            reset_metrics=request.reset_metrics,
+        )
+    except ValueError as exc:
+        if "already exists" in str(exc):
+            raise HTTPException(status.HTTP_409_CONFLICT, detail=str(exc)) from exc
+        raise
     if info is None:
         raise HTTPException(
-            status.HTTP_404_NOT_FOUND, detail="Source conversation not found"
+            status.HTTP_404_NOT_FOUND,
+            detail="Source conversation not found",
         )
     return info

openhands-agent-server/openhands/agent_server/conversation_service.py

@@ -674,10 +674,19 @@ async def fork_conversation(
         so the forked conversation is fully independent from the source.
 
         Returns ``None`` when *source_id* does not exist.
+
+        Raises:
+            ValueError: If *fork_id* is already taken by an active
+                conversation.
         """
         if self._event_services is None:
             raise ValueError("inactive_service")
 
+        # Reject duplicate fork IDs early to avoid clobbering an active
+        # conversation or leaking an EventService reference.
+        if fork_id is not None and fork_id in self._event_services:
+            raise ValueError(f"Conversation with id {fork_id} already exists")
+
         source_service = self._event_services.get(source_id)
         if source_service is None:
             return None
@@ -705,7 +714,14 @@ async def fork_conversation(
             agent=fork_agent,
             workspace=fork_workspace,
         )
-        fork_event_service = await self._start_event_service(fork_stored)
+        # If the service fails to start, clean up the orphaned persistence
+        # directory so we don't leave stale state on disk.
+        fork_dir = self.conversations_dir / fork_conv_id.hex
+        try:
+            fork_event_service = await self._start_event_service(fork_stored)
+        except Exception:
+            safe_rmtree(fork_dir)
+            raise
 
         state = await fork_event_service.get_state()
         return _compose_conversation_info_v1(fork_event_service.stored, state)

openhands-sdk/openhands/sdk/conversation/impl/local_conversation.py

@@ -348,54 +348,63 @@ def fork(
                 self.agent.model_dump(context={"expose_secrets": True}),
             )
 
-        # Determine persistence_dir for the fork
-        source_persistence = self._state.persistence_dir
-        fork_persistence: str | None = None
-        if source_persistence is not None:
-            source_path = Path(source_persistence)
-            fork_persistence = str(source_path.parent / fork_id.hex)
-
-        # Build the fork conversation (empty – no events yet)
-        fork_conv = LocalConversation(
-            agent=fork_agent,
-            workspace=self.workspace,
-            plugins=self._plugin_specs,
-            persistence_dir=fork_persistence,
-            conversation_id=fork_id,
-            max_iteration_per_run=self.max_iteration_per_run,
-            stuck_detection=self._stuck_detector is not None,
-            visualizer=type(self._visualizer) if self._visualizer else None,
-            delete_on_close=self.delete_on_close,
-            tags=tags,
-        )
+        # Hold the state lock while reading mutable state from the source
+        # conversation to avoid torn reads if run() is executing concurrently.
+        with self._state:
+            # Determine persistence_dir for the fork.
+            # Pass the *base* directory only — __init__ calls
+            # get_persistence_dir() which appends the conversation ID hex,
+            # so we must not do that here.
+            source_persistence = self._state.persistence_dir
+            fork_persistence: str | None = None
+            if source_persistence is not None:
+                source_path = Path(source_persistence)
+                fork_persistence = str(source_path.parent)
+
+            # Build the fork conversation (empty – no events yet)
+            fork_conv = LocalConversation(
+                agent=fork_agent,
+                workspace=self.workspace,
+                plugins=self._plugin_specs,
+                persistence_dir=fork_persistence,
+                conversation_id=fork_id,
+                max_iteration_per_run=self.max_iteration_per_run,
+                stuck_detection=self._stuck_detector is not None,
+                visualizer=type(self._visualizer) if self._visualizer else None,
+                delete_on_close=self.delete_on_close,
+                tags=tags,
+            )
 
-        # Deep-copy events from source → fork so the source stays immutable.
-        for event in self._state.events:
-            fork_conv._state.events.append(event.model_copy(deep=True))
-
-        # Copy runtime state that accumulated during the source conversation.
-        # activated_knowledge_skills is list[str] – strings are immutable so a
-        # shallow list copy is sufficient. agent_state can hold arbitrary
-        # mutable values, so deep-copy it.
-        fork_conv._state.activated_knowledge_skills = list(
-            self._state.activated_knowledge_skills
-        )
-        fork_conv._state.agent_state = copy.deepcopy(self._state.agent_state)
+            # Deep-copy events from source → fork so the source stays
+            # immutable.
+            for event in self._state.events:
+                fork_conv._state.events.append(event.model_copy(deep=True))
+
+            # Copy runtime state that accumulated during the source
+            # conversation. activated_knowledge_skills is list[str] – strings
+            # are immutable so a shallow list copy is sufficient.
+            # agent_state can hold arbitrary mutable values, so deep-copy it.
+            fork_conv._state.activated_knowledge_skills = list(
+                self._state.activated_knowledge_skills
+            )
+            fork_conv._state.agent_state = copy.deepcopy(self._state.agent_state)
+
+            # Copy title via tags if provided
+            if title is not None:
+                fork_conv._state.tags = {
+                    **fork_conv._state.tags,
+                    "title": title,
+                }
 
-        # Copy title via tags if provided
-        if title is not None:
-            fork_conv._state.tags = {
-                **fork_conv._state.tags,
-                "title": title,
-            }
+            # Reset or copy metrics
+            if not reset_metrics:
+                fork_conv._state.stats = self._state.stats.model_copy(deep=True)
 
-        # Reset or copy metrics
-        if not reset_metrics:
-            fork_conv._state.stats = self._state.stats.model_copy(deep=True)
+            event_count = len(self._state.events)
 
         logger.info(
             f"Forked conversation {self.id} → {fork_id} "
-            f"({len(self._state.events)} events copied, "
+            f"({event_count} events copied, "
             f"reset_metrics={reset_metrics})"
         )
         return fork_conv

tests/agent_server/test_conversation_router.py

@@ -1774,3 +1774,26 @@ def test_fork_conversation_not_found(
         assert response.status_code == 404
     finally:
         client.app.dependency_overrides.clear()
+
+
+def test_fork_conversation_duplicate_id_returns_409(
+    client, mock_conversation_service, sample_conversation_id
+):
+    """Test fork returns 409 when the requested fork ID already exists."""
+    mock_conversation_service.fork_conversation.side_effect = ValueError(
+        f"Conversation with id {sample_conversation_id} already exists"
+    )
+
+    client.app.dependency_overrides[get_conversation_service] = (
+        lambda: mock_conversation_service
+    )
+
+    try:
+        response = client.post(
+            f"/api/conversations/{sample_conversation_id}/fork",
+            json={"id": str(sample_conversation_id)},
+        )
+
+        assert response.status_code == 409
+    finally:
+        client.app.dependency_overrides.clear()

tests/sdk/conversation/local/test_fork.py

@@ -2,6 +2,7 @@
 
 import tempfile
 import uuid
+from pathlib import Path
 
 import pytest
 from pydantic import SecretStr
@@ -191,3 +192,61 @@ def test_fork_event_deep_copy_isolation():
         assert fork_evt.llm_message.content[0].text == "original"  # type: ignore[union-attr]
         fork_evt.llm_message.content[0].text = "mutated"  # type: ignore[union-attr]
         assert src_evt.llm_message.content[0].text == "original"  # type: ignore[union-attr]
+
+
+def test_fork_persistence_path_no_doubling():
+    """Fork persistence dir must be a sibling of source, not nested inside it.
+
+    Regression test: fork() previously computed the persistence path with
+    the conversation hex appended, but __init__ also appends it via
+    get_persistence_dir(), leading to /base/FORK_HEX/FORK_HEX.
+    """
+    with tempfile.TemporaryDirectory() as tmpdir:
+        src = Conversation(agent=_agent(), persistence_dir=tmpdir, workspace=tmpdir)
+        fork = src.fork()
+
+        assert src._state.persistence_dir is not None
+        assert fork._state.persistence_dir is not None
+        src_path = Path(src._state.persistence_dir)
+        fork_path = Path(fork._state.persistence_dir)
+
+        # Both should live directly under the same base directory
+        assert src_path.parent == fork_path.parent
+        # The fork dir should be <base>/<fork_id_hex>, not doubled
+        assert fork_path.name == fork.id.hex
+
+
+def test_fork_persisted_events_survive_reload():
+    """Events persisted by fork() should be loadable from the fork dir.
+
+    This validates the path-doubling fix end-to-end: if the fork wrote
+    events to the wrong directory, resuming from the correct path would
+    see zero events.
+    """
+    # Event IDs must be hex+dash, ≥8 chars to match EVENT_NAME_RE.
+    evt_id_1 = uuid.uuid4().hex
+    evt_id_2 = uuid.uuid4().hex
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        src = Conversation(agent=_agent(), persistence_dir=tmpdir, workspace=tmpdir)
+        src.state.events.append(_msg(evt_id_1, "hello"))
+        src.state.events.append(_msg(evt_id_2, "world"))
+
+        fork = src.fork()
+        fork_id = fork.id
+
+        # The fork should have the events in-memory
+        assert len(fork.state.events) == 2
+
+        # Close the fork to flush persistence, then reopen from disk
+        fork.close()
+
+        resumed = Conversation(
+            agent=_agent(),
+            persistence_dir=tmpdir,
+            workspace=tmpdir,
+            conversation_id=fork_id,
+        )
+        resumed_ids = [e.id for e in resumed.state.events]
+        assert evt_id_1 in resumed_ids
+        assert evt_id_2 in resumed_ids