fix(agent-server): base_state.json as single source of truth for the agent (end meta.json duplication)

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

@@ -671,6 +671,16 @@ def _load_persisted_state_sync(
             base_state_file.read_text(), context=context
         )
 
+    def _agent_from_base_state(self, conversation_id: UUID) -> AgentBase | None:
+        """Return the persisted agent from ``base_state.json`` (its single source
+        of truth), or ``None`` if there is no persisted state yet.
+
+        Used by cold-path checks (e.g. codex-agent detection) that used to read
+        the agent from ``meta.json`` before the agent was removed from it.
+        """
+        state = self._load_persisted_state_sync(conversation_id)
+        return state.agent if state is not None else None
+
     def _children_index(self) -> dict[UUID, list[UUID]]:
         """Reverse map parent_id -> child ids; rebuilt per call because the
         catalog is mutated from several places and a cache could go stale."""
@@ -756,11 +766,17 @@ async def _has_local_codex_credential(self) -> bool:
     async def _resolve_credential_bindings(
         self,
         stored: StoredConversation,
+        agent: AgentBase | None = None,
     ) -> dict[str, VersionedCredentialBinding]:
+        # The agent no longer lives on ``stored`` (meta.json). Callers pass the
+        # agent explicitly (the new-conversation request agent, or the live
+        # agent); otherwise fall back to the persisted base_state.json agent.
+        if agent is None:
+            agent = self._agent_from_base_state(stored.id)
         bindings = self._credential_bindings.pop(stored.id, {})
         if (
             CODEX_AUTH_SECRET_NAME not in bindings
-            and self._is_codex_agent(stored.agent)
+            and self._is_codex_agent(agent)
             and await self._has_local_codex_credential()
         ):
             assert self.secrets_store is not None
@@ -1208,8 +1224,13 @@ async def _start_conversation(
                     existing_event_service is not None
                     and existing_event_service.is_open()
                 ):
+                    existing_agent = (
+                        existing_event_service._conversation.agent
+                        if existing_event_service._conversation is not None
+                        else None
+                    )
                     if (
-                        self._is_codex_agent(existing_event_service.stored.agent)
+                        self._is_codex_agent(existing_agent)
                         and CODEX_AUTH_SECRET_NAME
                         not in existing_event_service.credential_bindings
                     ):
@@ -1261,7 +1282,7 @@ async def _start_conversation(
                         f"Persisted conversation {conversation_id} has no record"
                     )
                 managed_codex_credential = self._is_codex_agent(
-                    existing_record.stored.agent
+                    self._agent_from_base_state(conversation_id)
                 ) and (
                     CODEX_AUTH_SECRET_NAME
                     in self._credential_bindings.get(conversation_id, {})
@@ -1452,6 +1473,10 @@ async def _start_conversation(
             exclude={"agent_profile_id", "agent_launch_additions"},
         )
 
+        # The agent is persisted to base_state.json (not meta.json), so it is
+        # passed to _start_event_service separately. Default to request.agent.
+        new_agent: AgentBase = request.agent
+
         # If secrets_encrypted=True, the agent's secrets (e.g., LLM api_key) are
         # cipher-encrypted and need decryption during model validation. Pass the
         # cipher in the validation context so validate_secret() can decrypt them.
@@ -1473,15 +1498,26 @@ async def _start_conversation(
                 },
                 context={"cipher": self.cipher},
             )
+            # Decrypt the agent's secrets too (it no longer rides on `stored`).
+            # Re-validate the serialized agent with the cipher context so
+            # validate_secret() decrypts LLM api_key, MCP env, etc.
+            agent_cls = type(request.agent)
+            new_agent = agent_cls.model_validate(
+                request_data["agent"], context={"cipher": self.cipher}
+            )
         else:
             stored = StoredConversation(
                 id=conversation_id,
                 launched_agent_profile=launched_agent_profile,
                 **request_data,
             )
         async with self._lifecycle_lock:
+            # New conversation: the agent is written to base_state.json (its
+            # single source of truth), not to meta.json. Pass it explicitly.
+            # ``new_agent`` is ``request.agent`` (decrypted when the request was
+            # secrets_encrypted).
             event_service = await self._start_event_service(
-                stored, is_new_conversation=True
+                stored, is_new_conversation=True, agent=new_agent
             )
         initial_message = request.initial_message
         if initial_message:
@@ -1735,9 +1771,12 @@ async def fork_conversation(
         # fork-specific fields. Without this, e.g. a fork of a client-tool
         # conversation would lose ``client_tools`` in meta.json and be unable
         # to re-register its tools after a server restart.
+        # Note: the agent is NOT stored in meta.json (StoredConversation) — the
+        # fork's agent is already persisted to the fork's base_state.json by
+        # ``source_conversation.fork`` above. It is passed to
+        # ``_start_event_service`` via ``agent=`` for the new-conversation path.
         fork_overrides: dict[str, Any] = {
             "id": fork_conv_id,
-            "agent": fork_agent,
             "workspace": fork_workspace,
             "title": title,
             "created_at": utc_now(),
@@ -1756,7 +1795,7 @@ async def fork_conversation(
         try:
             async with self._lifecycle_lock:
                 fork_event_service = await self._start_event_service(
-                    fork_stored, is_new_conversation=True
+                    fork_stored, is_new_conversation=True, agent=fork_agent
                 )
         except Exception:
             safe_rmtree(fork_dir)
@@ -2023,16 +2062,26 @@ def get_instance(cls, config: Config) -> "ConversationService":
         )
 
     async def _start_event_service(
-        self, stored: StoredConversation, *, is_new_conversation: bool = False
+        self,
+        stored: StoredConversation,
+        *,
+        is_new_conversation: bool = False,
+        agent: AgentBase | None = None,
     ) -> EventService:
         event_services = self._event_services
         if event_services is None:
             raise ValueError("inactive_service")
 
-        credential_bindings = await self._resolve_credential_bindings(stored)
+        # ``agent`` is supplied for a NEW conversation (meta.json no longer
+        # carries it). On resume it is ``None`` and both the credential check
+        # and EventService read the agent from base_state.json.
+        credential_bindings = await self._resolve_credential_bindings(
+            stored, agent=agent
+        )
         event_service = EventService(
             stored=stored,
             conversations_dir=self.conversations_dir,
+            agent=agent,
             cipher=self.cipher,
             mcp_tool_provider=self.mcp_tool_provider,
             credential_bindings=credential_bindings,
@@ -2138,11 +2187,15 @@ async def _maybe_subscribe_telemetry(
             if factory is None:
                 return
 
+            live_conversation = getattr(event_service, "_conversation", None)
+            live_agent = (
+                live_conversation.agent if live_conversation is not None else None
+            )
             subscriber = TelemetrySubscriber(
                 conversation_id=stored.id,
                 sink=sink,
                 factory=factory,
-                context=_build_telemetry_context(stored, factory),
+                context=_build_telemetry_context(stored, factory, agent=live_agent),
             )
             await event_service.subscribe_to_events(subscriber)
             if is_new_conversation:
@@ -2152,14 +2205,21 @@ async def _maybe_subscribe_telemetry(
 
 
 def _build_telemetry_context(
-    stored: StoredConversation, factory: DiagnosticEventFactory
+    stored: StoredConversation,
+    factory: DiagnosticEventFactory,
+    agent: AgentBase | None = None,
 ) -> ConversationTelemetryContext:
     """Reduce a stored conversation to its sanitized telemetry facts.
 
     Every read is defensive: a shape change upstream should degrade a property
     to ``unknown``, never raise into conversation startup.
+
+    The agent is no longer stored on meta.json; callers pass the live/persisted
+    agent explicitly. ``getattr(stored, "agent", None)`` remains as a defensive
+    fallback for older callers.
     """
-    agent = getattr(stored, "agent", None)
+    if agent is None:
+        agent = getattr(stored, "agent", None)
     llm = getattr(agent, "llm", None)
 
     workspace = getattr(stored, "workspace", None)

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

@@ -115,6 +115,11 @@ class EventService:
 
     stored: StoredConversation
     conversations_dir: Path
+    # Agent for a NEW conversation. meta.json (``stored``) no longer carries the
+    # agent — base_state.json is its single source of truth — so the creating
+    # caller passes it here. On resume this is ``None`` and the agent is loaded
+    # from base_state.json.
+    agent: AgentBase | None = None
     cipher: Cipher | None = None
     mcp_tool_provider: MCPToolProvider | None = None
     credential_bindings: dict[str, VersionedCredentialBinding] = field(
@@ -179,17 +184,13 @@ async def save_meta(self):
             )
 
     def _without_stored_secret(self, secret_name: str) -> StoredConversation:
+        # meta.json (StoredConversation) no longer carries the agent, so there is
+        # no agent_context secret to scrub here — only the stored secrets map.
+        # The agent's own secret scrub happens on base_state.json (see
+        # _scrub_persisted_credentials).
         secrets = dict(self.stored.secrets)
         secrets.pop(secret_name, None)
-        return self.stored.model_copy(
-            update={
-                "secrets": secrets,
-                "agent": _without_agent_context_secret(
-                    self.stored.agent,
-                    secret_name,
-                ),
-            }
-        )
+        return self.stored.model_copy(update={"secrets": secrets})
 
     async def _scrub_persisted_credentials(
         self,
@@ -972,10 +973,24 @@ async def start(self):
         working_dir = Path(workspace.working_dir)
         working_dir.mkdir(parents=True, exist_ok=True)
         self._ensure_workspace_is_git_repo(working_dir)
-        agent_cls = type(self.stored.agent)
-        agent = agent_cls.model_validate(
-            self.stored.agent.model_dump(context={"expose_secrets": True}),
-        )
+        # base_state.json is the single source of truth for the agent. On resume
+        # (base_state exists) pass ``agent=None`` so LocalConversation keeps the
+        # persisted agent. On a new conversation the creating caller supplied the
+        # agent via ``self.agent``; deep-copy it (expose_secrets) so the running
+        # agent is independent of the caller's object.
+        base_state_exists = (self.conversation_dir / BASE_STATE).exists()
+        if base_state_exists:
+            agent: AgentBase | None = None
+        else:
+            if self.agent is None:
+                raise ValueError(
+                    "Cannot start a new conversation without an agent: no "
+                    "base_state.json to resume and no agent was provided."
+                )
+            agent_cls = type(self.agent)
+            agent = agent_cls.model_validate(
+                self.agent.model_dump(context={"expose_secrets": True}),
+            )
 
         # Create LocalConversation with plugins and hook_config.
         # Plugins are loaded lazily on first run()/send_message() call.
@@ -987,16 +1002,18 @@ async def start(self):
             self._pub_sub, loop=asyncio.get_running_loop()
         )
 
-        # Only wire token streaming for agents that can actually emit token
-        # callbacks. SDK LLM agents need stream=True, while ACP agents emit
-        # AgentMessageChunk text through their bridge without exposing an LLM.
-        streaming_enabled = isinstance(agent, ACPAgent) or any(
-            llm.stream for llm in agent.get_all_llms()
-        )
-        logger.debug(
-            "Token streaming: %s",
-            "enabled" if streaming_enabled else "disabled (no LLM has stream=True)",
-        )
+        # Token streaming is wired only for agents that can actually emit token
+        # callbacks (SDK LLM agents with stream=True, or ACP agents). For a NEW
+        # conversation the agent is known here, so decide now. On RESUME the
+        # agent is loaded from base_state.json during construction, so defer the
+        # decision until after (see the post-construction block below).
+        def _agent_can_stream(a: AgentBase) -> bool:
+            return isinstance(a, ACPAgent) or any(
+                llm.stream for llm in a.get_all_llms()
+            )
+
+        streaming_enabled = _agent_can_stream(agent) if agent is not None else True
+        streaming_decided = agent is not None
 
         def _publish_stream_delta(
             content: str | None = None,
@@ -1059,6 +1076,17 @@ def _token_streaming_callback(chunk: LLMStreamChunk | str) -> None:
 
         conversation.set_confirmation_policy(self.stored.confirmation_policy)
         conversation.set_security_analyzer(self.stored.security_analyzer)
+        # On resume the agent was unknown at construction time (loaded from
+        # base_state.json), so decide token streaming now and disable it when the
+        # resolved agent can't emit token callbacks.
+        if not streaming_decided:
+            streaming_enabled = _agent_can_stream(conversation.agent)
+            logger.debug(
+                "Token streaming: %s",
+                "enabled" if streaming_enabled else "disabled (no LLM has stream=True)",
+            )
+            if not streaming_enabled:
+                conversation._on_token = None
         self._conversation = conversation
         if isinstance(conversation.agent, ACPAgent):
             for secret_name, binding in self.credential_bindings.items():
@@ -1621,13 +1649,13 @@ async def switch_acp_model(self, model: str) -> None:
 
         For a conversation that has already started, runs the (blocking)
         protocol-level ``session/set_model`` round-trip in a worker thread; for
-        one not yet run, the SDK defers the switch (persist-only). Either way it
-        mirrors the new model into ``meta.json`` so the switch survives an
-        agent-server restart: ``start()`` rebuilds the agent from
-        ``self.stored.agent`` and ``ConversationState.create()`` copies that over
-        the persisted base_state.json on resume. Only ``acp_model`` needs
-        updating — ``model_post_init`` re-derives the sentinel ``llm.model`` on
-        reload.
+        one not yet run, the SDK defers the switch (persist-only). Either way the
+        switched model is persisted as the authoritative value in
+        ``base_state.json``: ``LocalConversation.switch_acp_model`` sets
+        ``state.agent`` to an agent copy carrying the new ``acp_model``, which the
+        autosave path writes to base_state. On resume the agent is rebuilt from
+        base_state (the single source of truth), so no ``meta.json`` mirror is
+        needed.
         """
         if self._conversation is None:
             # Match the inactive-service convention of the other event-service
@@ -1637,10 +1665,6 @@ async def switch_acp_model(self, model: str) -> None:
             raise ValueError("inactive_service")
         loop = asyncio.get_running_loop()
         await loop.run_in_executor(None, self._conversation.switch_acp_model, model)
-        self.stored = self.stored.model_copy(
-            update={"agent": self.stored.agent.model_copy(update={"acp_model": model})}
-        )
-        await self.save_meta()
 
     async def close(self):
         self._closing = True

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

@@ -13,6 +13,7 @@
 from openhands.sdk.conversation.conversation_stats import ConversationStats
 from openhands.sdk.conversation.request import (  # re-export for backward compat
     ACPEnabledAgent as ACPEnabledAgent,
+    ConversationConfig as ConversationConfig,
     SendMessageRequest as SendMessageRequest,
     StartConversationRequest as StartConversationRequest,
 )
@@ -73,15 +74,17 @@ class EventSortOrder(StrEnum):
     TIMESTAMP_DESC = "TIMESTAMP_DESC"
 
 
-class StoredConversation(StartConversationRequest):
+class StoredConversation(ConversationConfig):
     """Stored details about a conversation.
 
-    Extends StartConversationRequest with server-assigned fields.
+    Extends :class:`ConversationConfig` (the agent-less shared config) with
+    server-assigned fields. It deliberately does NOT carry the ``agent``: the
+    single source of truth for the agent / runtime state is
+    ``ConversationState`` persisted to ``base_state.json``. Because
+    ``StoredConversation`` is not a ``StartConversationRequest``, the agent
+    cannot silently re-appear in ``meta.json``.
     """
 
-    # agent_profile_id is resolved into launched_agent_profile at creation; exclude from
-    # the persistence payload so it does not re-appear in meta.json.
-    agent_profile_id: UUID | None = Field(default=None, exclude=True)
     required_runtime_credential_bindings: set[str] = Field(default_factory=set)
 
     id: OpenHandsUUID