diff --git a/agent/admin.py b/agent/admin.py index 582386e1..83f2b551 100644 --- a/agent/admin.py +++ b/agent/admin.py @@ -278,6 +278,20 @@ class AgentAdminService: except OSError as e: logger.warning(f"[AgentAdmin] Could not create own knowledge for {destination}: {e}") + @staticmethod + def _make_own_skills(destination: Path) -> None: + """Give a brand-new Agent its own skill set (opt out of shared). + + Presence of the directory is what opts an Agent out of the shared copy, + so an empty ``skills/`` is enough: the Agent then starts with no shared + skills and installs its own. + """ + sdir = destination / "skills" + try: + sdir.mkdir(parents=True, exist_ok=True) + except OSError as e: + logger.warning(f"[AgentAdmin] Could not create own skills for {destination}: {e}") + @staticmethod def _clone_persona(source: Path, destination: Path) -> None: """Copy how an Agent behaves, and nothing else. @@ -337,10 +351,13 @@ class AgentAdminService: skills: Optional[Iterable[str]] = None, knowledge: Optional[Iterable[str]] = None, knowledge_mode: str = None, + skill_mode: str = None, revision: str = None, ) -> Dict: if knowledge_mode not in (None, "shared", "own"): raise AgentAdminError("knowledge mode must be 'shared' or 'own'") + if skill_mode not in (None, "shared", "own"): + raise AgentAdminError("skill mode must be 'shared' or 'own'") with self._lock: settings = self._load() registry = self._registry(settings) @@ -388,6 +405,8 @@ class AgentAdminService: self._seed_name(workspace, name) if knowledge_mode == "own": self._make_own_knowledge(destination) + if skill_mode == "own": + self._make_own_skills(destination) profile = AgentProfile( id=agent_id, diff --git a/agent/chat/service.py b/agent/chat/service.py index b11cb516..f80f3972 100644 --- a/agent/chat/service.py +++ b/agent/chat/service.py @@ -35,6 +35,8 @@ class ChatService: channel_type: str = "", agent_id: str = None, request_id: str = None, # noqa: RUF013 + speaker_agent_id: str = None, + members: list = None, ): """ Run the agent for *query* and stream results back via *send_chunk_fn*. @@ -46,21 +48,16 @@ class ChatService: :param session_id: session identifier for agent isolation :param send_chunk_fn: callable(chunk_data: dict) to send a streaming chunk :param channel_type: source channel (e.g. "web", "feishu") for persistence - :param agent_id: selected agent profile; defaults to the configured default + :param agent_id: agent that owns the conversation; defaults to the configured default :param request_id: per-request cancellation key; defaults to session scope + :param speaker_agent_id: teammate addressed for this turn; it answers in + the owner's conversation, so the transcript stays in one place + :param members: roster of the conversation (teammate ids). When given it + is authoritative and reconciled onto the session, like a team channel """ + # The conversation belongs to ``resolved_agent_id`` (its owner); only the + # voice answering this turn may differ. Same model as agent_reply. resolved_agent_id = self.agent_bridge._resolve_agent_id(agent_id) - agent = self.agent_bridge.get_agent( - session_id=session_id, agent_id=resolved_agent_id - ) - if agent is None: - raise RuntimeError("Failed to initialise agent for the session") - - # Pass context metadata to model for downstream API requests - if hasattr(agent, 'model'): - agent.model.channel_type = channel_type or "" - agent.model.session_id = session_id or "" - agent.model.agent_id = resolved_agent_id # Build a context so context-aware tools (e.g. scheduler) can resolve the # receiver/session. This streaming path bypasses agent_bridge.agent_reply, @@ -68,6 +65,45 @@ class ChatService: context = self._build_context( query, session_id, channel_type, resolved_agent_id ) + speaker_id = resolved_agent_id + model_query = query + is_team = False + if speaker_agent_id or members is not None: + if members is not None: + context["members"] = list(members) + if speaker_agent_id: + context["speaker_agent_id"] = speaker_agent_id + self.agent_bridge._seed_team_members(session_id, resolved_agent_id, context) + speaker_id = self.agent_bridge._resolve_speaker(resolved_agent_id, context) + is_team = speaker_id != resolved_agent_id or self._has_team(session_id, resolved_agent_id) + if speaker_agent_id: + # What the model is asked once the address has been acted on + # (also when the owner itself was named); the transcript keeps + # the verbatim query. + model_query = self.agent_bridge._strip_address(query, speaker_id) + agent = self.agent_bridge.get_agent( + session_id=session_id, + agent_id=speaker_id, + host_agent_id=resolved_agent_id, + ) + else: + agent = self.agent_bridge.get_agent( + session_id=session_id, agent_id=resolved_agent_id + ) + if agent is None: + raise RuntimeError("Failed to initialise agent for the session") + if is_team: + # One transcript per team conversation: reload it with author labels + # so this speaker sees the turns others spoke since it last ran. + self.agent_bridge._sync_shared_transcript(agent, session_id, resolved_agent_id) + self._send_speaker(send_chunk_fn, speaker_id) + + # Pass context metadata to model for downstream API requests + if hasattr(agent, 'model'): + agent.model.channel_type = channel_type or "" + agent.model.session_id = session_id or "" + agent.model.agent_id = speaker_id + self._attach_context_aware_tools(agent, context) # Mark this session as mid-run so the self-evolution idle scan does not @@ -305,7 +341,7 @@ class ChatService: f"session={session_id}" ) return - response = executor.run_stream(query) + response = executor.run_stream(model_query) except Exception: # If executor cleared messages (context overflow), sync back if len(executor.messages) == 0: @@ -381,13 +417,23 @@ class ChatService: agent.messages = list(executor.messages) # Persist new messages to SQLite so they survive restarts and - # can be queried via the HISTORY interface. + # can be queried via the HISTORY interface. The store is the owner's: + # a guest speaker writes into the shared transcript, stamped as author. if new_messages: + workspace_root = agent.workspace_dir + if is_team: + new_messages = self.agent_bridge._attribute_to_speaker( + new_messages, speaker_id + ) + new_messages = self.agent_bridge._strip_speaker_prefix_from_messages( + new_messages + ) + workspace_root = self._owner_workspace(resolved_agent_id, agent) self._persist_messages( session_id, list(new_messages), channel_type, - workspace_root=agent.workspace_dir, + workspace_root=workspace_root, ) # Store executor reference for files_to_send access @@ -428,6 +474,40 @@ class ChatService: ctx["agent_id"] = agent_id return ctx + @staticmethod + def _has_team(session_id: str, host_agent_id: str) -> bool: + """Whether the session has teammates on it (a shared conversation).""" + if not session_id: + return False + try: + from agent.workspace import session_prefs + return bool(session_prefs.get_prefs(session_id, host_agent_id).get("members")) + except Exception: + return False + + def _send_speaker(self, send_chunk_fn, speaker_id: str) -> None: + """Tell the client who is answering this turn, so a shared conversation + can attribute the reply to the right agent as it streams.""" + try: + profile = self.agent_bridge.agent_registry.get(speaker_id, require_enabled=False) + send_chunk_fn({ + "chunk_type": "speaker", + "agent_id": profile.id, + "name": profile.name, + "avatar": profile.avatar or "", + }) + except Exception as e: + logger.debug(f"[ChatService] speaker chunk skipped: {e}") + + def _owner_workspace(self, owner_agent_id: str, agent) -> str: + """Workspace whose store holds the conversation: the owner's.""" + try: + return self.agent_bridge.agent_registry.get( + owner_agent_id, require_enabled=False + ).workspace + except Exception: + return agent.workspace_dir + def _attach_context_aware_tools(self, agent, context): """Attach the current context to tools that need turn metadata.""" try: diff --git a/agent/registry.py b/agent/registry.py index eddb283d..72752b88 100644 --- a/agent/registry.py +++ b/agent/registry.py @@ -18,6 +18,10 @@ from common.utils import expand_path _AGENT_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,63}$") +# Id of the built-in single agent, and the reserved alias by which the default +# agent can always be addressed (see AgentRegistry.get_addressed). +DEFAULT_AGENT_ALIAS = "default" + class AgentRegistryError(ValueError): """Raised when agent configuration is invalid.""" @@ -192,8 +196,24 @@ class AgentRegistry: raw_agents = settings.get("agents") if raw_agents is None or raw_agents == []: - profile = AgentProfile(id="default", name="CowAgent", workspace=instance_root) - return cls([profile], "default") + # Single-agent install: the sole agent owns the instance root. Its id + # is normally the built-in "default", but a provisioner may inject one + # so the id is stable and meaningful from first boot; an invalid or + # absent value falls back to "default" and never breaks startup. + builtin_id = settings.get("default_agent_id") + # A value injected via env may arrive as a non-string (e.g. an + # all-digit id parsed as int), so coerce before validating. + builtin_id = str(builtin_id).strip() if builtin_id is not None else "" + if not builtin_id or not _AGENT_ID_RE.match(builtin_id): + builtin_id = DEFAULT_AGENT_ALIAS + builtin_name = settings.get("default_agent_name") + builtin_name = str(builtin_name).strip() if builtin_name is not None else "" + profile = AgentProfile( + id=builtin_id, + name=builtin_name or "CowAgent", + workspace=instance_root, + ) + return cls([profile], builtin_id) if not isinstance(raw_agents, list): raise AgentRegistryError("agents must be a list") @@ -261,6 +281,20 @@ class AgentRegistry: raise AgentRegistryError(f"agent '{resolved_id}' is disabled") return profile + def get_addressed(self, agent_id: Optional[str], require_enabled: bool = True) -> AgentProfile: + """The agent an inbound request addresses by id. + + Like ``get``, except that the reserved id ``"default"`` always means + the default agent, whatever id it was actually given. Remote callers + can therefore address the default agent without knowing its real id. + Any other unknown id still raises, so a request is never silently + answered by a different agent. + """ + requested = (agent_id or "").strip() + if requested == DEFAULT_AGENT_ALIAS: + requested = "" + return self.get(requested or None, require_enabled=require_enabled) + def get_or_default(self, agent_id: Optional[str]) -> AgentProfile: with self._lock: profile = self._profiles.get(agent_id) if agent_id else None diff --git a/agent/routing.py b/agent/routing.py index 50e0743e..a54cf55b 100644 --- a/agent/routing.py +++ b/agent/routing.py @@ -42,7 +42,7 @@ class AgentRouter: and nothing in the conversation would reveal the substitution. """ try: - return self.registry.get(agent_id).id + return self.registry.get_addressed(agent_id).id except Exception as exc: raise AgentUnavailableError( f"{source} selected agent {agent_id!r}, which is missing or disabled" diff --git a/common/cloud_client.py b/common/cloud_client.py index 653c7e40..def8b18f 100644 --- a/common/cloud_client.py +++ b/common/cloud_client.py @@ -37,6 +37,9 @@ chat_client: LinkAIClient CHANNEL_ACTIONS = {"channel_create", "channel_update", "channel_delete"} +AGENT_ACTIONS = {"agent_create", "agent_update", "agent_delete"} + + # channelType -> config key mapping for app credentials. # secret_key may be "" for single-token channels (e.g. telegram/discord). # For slack, appId carries bot_token and appSecret carries app_token. @@ -77,7 +80,12 @@ def _acting_user(user_id): class CloudClient(LinkAIClient): def __init__(self, api_key: str, channel, host: str = "", port=None): - super().__init__(api_key, host, port=port) + # Older base client versions do not accept a ``port`` kwarg; fall back + # to the positional signature so both old and new versions work. + try: + super().__init__(api_key, host, port=port) + except TypeError: + super().__init__(api_key, host) self.channel = channel self.client_type = channel.channel_type self.channel_mgr = None @@ -194,6 +202,10 @@ class CloudClient(LinkAIClient): self._dispatch_channel_action(action, config.get("data", {})) return + if action in AGENT_ACTIONS: + self._dispatch_agent_action(action, config.get("data", {})) + return + if config.get("enabled") != "Y": return @@ -302,6 +314,60 @@ class CloudClient(LinkAIClient): elif action == "channel_delete": self._handle_channel_delete(channel_type, data) + # ------------------------------------------------------------------ + # agent operations (add/remove an agent without a restart) + # ------------------------------------------------------------------ + def _dispatch_agent_action(self, action: str, data: dict): + agent_id = str(data.get("id") or data.get("agentId") or "").strip() + if not agent_id: + logger.warning(f"[CloudClient] Agent action '{action}' missing id, data={data}") + return + logger.info(f"[CloudClient] Agent action: {action}, id={agent_id}") + if action == "agent_create": + self._handle_agent_create(agent_id, data) + + def _handle_agent_create(self, agent_id: str, data: dict): + """Add a new agent and re-point the live runtime, so it can answer + without a restart. A no-op when agent support is unavailable.""" + name = str(data.get("name") or agent_id).strip() + model = data.get("model") + # Asset isolation for the new agent. Values are "own" (a private copy) + # or "shared" (draw on the shared library); unset keeps the default + # shared behaviour so existing callers are unaffected. + knowledge_mode = data.get("knowledge_mode") or data.get("knowledgeMode") + skill_mode = data.get("skill_mode") or data.get("skillMode") + try: + from agent.admin import get_agent_admin_service + except Exception as e: + logger.warning(f"[CloudClient] agent create unavailable: {e}") + return + try: + service = get_agent_admin_service() + service.create_agent( + agent_id=agent_id, + name=name, + knowledge_mode=knowledge_mode, + skill_mode=skill_mode, + ) + if model: + try: + service.update_agent(agent_id, model=model) + except Exception as e: + logger.warning(f"[CloudClient] set agent model failed: {e}") + self._reload_agents(service) + logger.info(f"[CloudClient] Agent '{agent_id}' created") + except Exception as e: + logger.error(f"[CloudClient] Failed to create agent '{agent_id}': {e}", exc_info=True) + + @staticmethod + def _reload_agents(service): + """Re-point the running runtime at the updated roster.""" + try: + from channel.web.web_channel import _reload_agent_runtime + _reload_agent_runtime(service) + except Exception as e: + logger.warning(f"[CloudClient] agent runtime reload skipped: {e}") + # ------------------------------------------------------------------ # per-instance channel operations (multi-instance path) # ------------------------------------------------------------------ @@ -673,15 +739,33 @@ class CloudClient(LinkAIClient): :return: response dict """ action = data.get("action", "") - payload = data.get("payload") + payload = data.get("payload") or {} logger.info(f"[CloudClient] on_memory: action={action}") - svc = self.memory_service + agent_id = payload.get("agent_id") or payload.get("agentId") + try: + svc = self._memory_service_for(agent_id) + except KeyError: + return self._agent_not_found(action, agent_id) if svc is None: return {"action": action, "code": 500, "message": "MemoryService not available", "payload": None} return svc.dispatch(action, payload) + def _memory_service_for(self, agent_id): + """A MemoryService bound to the requested agent's workspace. Falls back + to the process-wide default-agent service when no agent is requested, + so single-agent installs are unaffected.""" + workspace = self._agent_workspace(agent_id) + if workspace is None: + return self.memory_service + try: + from agent.memory.service import MemoryService + return MemoryService(workspace) + except Exception as e: + logger.error(f"[CloudClient] Failed to build MemoryService for agent: {e}") + return self.memory_service + # ------------------------------------------------------------------ # knowledge callback # ------------------------------------------------------------------ @@ -694,15 +778,34 @@ class CloudClient(LinkAIClient): :return: response dict """ action = data.get("action", "") - payload = data.get("payload") + payload = data.get("payload") or {} logger.info(f"[CloudClient] on_knowledge: action={action}") - svc = self.knowledge_service + agent_id = payload.get("agent_id") or payload.get("agentId") + try: + svc = self._knowledge_service_for(agent_id) + except KeyError: + return self._agent_not_found(action, agent_id) if svc is None: return {"action": action, "code": 500, "message": "KnowledgeService not available", "payload": None} return svc.dispatch(action, payload) + def _knowledge_service_for(self, agent_id): + """A KnowledgeService bound to the requested agent's workspace. Falls + back to the process-wide default-agent service when no agent is + requested, so single-agent installs are unaffected. An agent with its + own knowledge/ reads that; otherwise it transparently reads the shared one.""" + workspace = self._agent_workspace(agent_id) + if workspace is None: + return self.knowledge_service + try: + from agent.knowledge.service import KnowledgeService + return KnowledgeService(str(workspace)) + except Exception as e: + logger.error(f"[CloudClient] Failed to build KnowledgeService for agent: {e}") + return self.knowledge_service + # ------------------------------------------------------------------ # workspace callback # ------------------------------------------------------------------ @@ -721,12 +824,30 @@ class CloudClient(LinkAIClient): logger.info(f"[CloudClient] on_workspace: action={action}, path={payload.get('path', '')}") - svc = self.workspace_service + agent_id = payload.get("agent_id") or payload.get("agentId") + try: + svc = self._workspace_service_for(agent_id) + except KeyError: + return self._agent_not_found(action, agent_id) if svc is None: return {"action": action, "code": 500, "message": "WorkspaceService not available", "payload": None} return svc.dispatch(action, payload) + def _workspace_service_for(self, agent_id): + """A WorkspaceService rooted at the requested agent's workspace. Falls + back to the process-wide default-agent service when no agent is + requested, so single-agent installs are unaffected.""" + workspace = self._agent_workspace(agent_id) + if workspace is None: + return self.workspace_service + try: + from agent.workspace.service import WorkspaceService + return WorkspaceService(str(workspace)) + except Exception as e: + logger.error(f"[CloudClient] Failed to build WorkspaceService for agent: {e}") + return self.workspace_service + # ------------------------------------------------------------------ # chat callback # ------------------------------------------------------------------ @@ -745,10 +866,25 @@ class CloudClient(LinkAIClient): # Console user on whose behalf this runs; usage is attributed to them # instead of the account this deployment is registered under. user_id = payload.get("user_id") or data.get("user_id") + # Which agent should answer. Absent means the default agent, so + # single-agent installs keep working unchanged. + agent_id = payload.get("agent_id") or payload.get("agentId") + agent_id = self._resolve_chat_agent_id(agent_id) + # Shared conversation: the roster on it and the teammate addressed for + # this turn. Both optional; absent keeps the single-agent behaviour. + speaker_agent_id = self._resolve_optional_agent_id( + payload.get("speaker_agent_id") or payload.get("speakerAgentId") + ) + members = payload.get("members") + if isinstance(members, list): + members = [m for m in (self._resolve_optional_agent_id(x) for x in members) if m] + else: + members = None if not session_id.startswith("session_"): session_id = f"session_{session_id}" logger.info(f"[CloudClient] on_chat: session={session_id}, channel={channel_type}, " - f"user_id={user_id}, query={query[:80]}") + f"user_id={user_id}, agent_id={agent_id}, speaker={speaker_agent_id}, " + f"members={members}, query={query[:80]}") # Cancel / steer fast-path. These are NOT new agent turns — they act on # the run already in flight for this session. The web channel intercepts @@ -758,7 +894,7 @@ class CloudClient(LinkAIClient): stripped = (query or "").strip() steer_flag = bool(payload.get("steer")) if stripped == "/cancel": - handled = self._handle_cancel(session_id, send_chunk_fn) + handled = self._handle_cancel(session_id, send_chunk_fn, agent_id) if handled: return elif steer_flag or stripped.startswith("/steer"): @@ -767,10 +903,10 @@ class CloudClient(LinkAIClient): if stripped.startswith("/steer") else stripped ) - self._handle_steer(session_id, instruction, send_chunk_fn) + self._handle_steer(session_id, instruction, send_chunk_fn, agent_id) return - with _acting_user(user_id): + with _acting_user(user_id), self._chat_identity(agent_id, user_id, session_id): # Intercept cow/slash commands before the agent runs try: from plugins import PluginManager @@ -789,7 +925,98 @@ class CloudClient(LinkAIClient): raise RuntimeError("ChatService not available") svc.run(query=query, session_id=session_id, channel_type=channel_type, - send_chunk_fn=send_chunk_fn) + send_chunk_fn=self._aliasing_sender(send_chunk_fn), agent_id=agent_id, + speaker_agent_id=speaker_agent_id, members=members) + + def _aliasing_sender(self, send_chunk_fn): + """Report the default agent to remote callers by its reserved alias, + matching how they address it (see AgentRegistry.get_addressed).""" + def send(chunk): + if isinstance(chunk, dict) and chunk.get("chunk_type") == "speaker": + chunk = {**chunk, "agent_id": self._alias_agent_id(chunk.get("agent_id"))} + send_chunk_fn(chunk) + return send + + @staticmethod + def _alias_agent_id(agent_id): + """The default agent's id as seen from outside: the reserved alias.""" + try: + from agent.registry import DEFAULT_AGENT_ALIAS, get_agent_registry + if agent_id and agent_id == get_agent_registry().default_agent_id: + return DEFAULT_AGENT_ALIAS + except Exception: + pass + return agent_id + + def _resolve_optional_agent_id(self, agent_id): + """Map an addressed id (including the reserved ``"default"`` alias) to + the configured agent id, or None when absent or unknown. Unknown + teammates are dropped rather than failing the turn: the owner still + answers, which is what ``_resolve_speaker`` does for them anyway.""" + agent_id = str(agent_id).strip() if agent_id is not None else "" + if not agent_id: + return None + try: + from agent.registry import get_agent_registry + return get_agent_registry().get_addressed(agent_id, require_enabled=False).id + except Exception: + logger.warning(f"[CloudClient] unknown agent id ignored: {agent_id}") + return None + + def _resolve_chat_agent_id(self, agent_id): + """Validate a requested agent id, or fall back to the default agent + when none is requested. + + Returns None when no bridge is available yet so callers keep their + existing default-agent behaviour. A requested agent that is unknown or + disabled is an error: answering as the default agent instead would + silently mix up two agents' sessions and memory. + """ + agent_id = str(agent_id).strip() if agent_id is not None else "" + bridge = self._agent_bridge() + if bridge is None: + return agent_id or None + try: + return bridge.agent_router.resolve(explicit_agent_id=agent_id or None) + except Exception as e: + if agent_id: + raise RuntimeError(f"agent unavailable: {agent_id}") from e + logger.warning(f"[CloudClient] agent route fallback to default: {e}") + return None + + def _agent_workspace(self, agent_id): + """Resolve a requested agent id to its workspace root, or None to keep + the default-agent behaviour when no agent is requested. An agent that is + requested but unknown raises KeyError rather than silently serving the + default agent's data (see AgentRegistry.get_addressed for the ids that + are tolerated).""" + agent_id = str(agent_id).strip() if agent_id is not None else "" + if not agent_id: + return None + try: + from agent.registry import get_agent_registry + registry = get_agent_registry() + except Exception as e: + logger.warning(f"[CloudClient] agent registry unavailable, using default: {e}") + return None + return registry.get_addressed(agent_id, require_enabled=False).workspace + + @staticmethod + def _agent_not_found(action, agent_id): + return {"action": action, "code": 404, "message": f"agent not found: {agent_id}", "payload": None} + + @contextmanager + def _chat_identity(self, agent_id, user_id, session_id): + """Bind the ambient identity so downstream code resolves the right + agent workspace/memory. A no-op when identity support is absent.""" + try: + from common.runtime_identity import identity_scope + except Exception: + yield + return + with identity_scope(agent_id=agent_id, user_id=(str(user_id) if user_id else None), + session_id=session_id): + yield def _agent_bridge(self): try: @@ -799,7 +1026,7 @@ class CloudClient(LinkAIClient): logger.warning(f"[CloudClient] agent_bridge unavailable: {e}") return None - def _handle_cancel(self, session_id: str, send_chunk_fn) -> bool: + def _handle_cancel(self, session_id: str, send_chunk_fn, agent_id: str = None) -> bool: """Abort the in-flight run for this session. Returns True if it was our command to handle (always True once matched), regardless of whether a run was actually running.""" @@ -808,7 +1035,7 @@ class CloudClient(LinkAIClient): if bridge is not None: try: from agent.protocol import get_cancel_registry - key = bridge.scoped_session_key(session_id) + key = bridge.scoped_session_key(session_id, agent_id) cancelled = get_cancel_registry().cancel_session(key) except Exception as e: logger.warning(f"[CloudClient] cancel failed: {e}") @@ -817,7 +1044,8 @@ class CloudClient(LinkAIClient): send_chunk_fn({"chunk_type": "content", "delta": msg, "segment_id": 0}) return True - def _handle_steer(self, session_id: str, instruction: str, send_chunk_fn) -> None: + def _handle_steer(self, session_id: str, instruction: str, send_chunk_fn, + agent_id: str = None) -> None: """Inject a mid-run instruction into this session's active run.""" if not instruction: send_chunk_fn({"chunk_type": "content", @@ -827,7 +1055,7 @@ class CloudClient(LinkAIClient): status_val = None if bridge is not None: try: - result = bridge.steer_session(session_id, instruction) + result = bridge.steer_session(session_id, instruction, agent_id) status_val = getattr(getattr(result, "status", None), "value", None) or str(result) except Exception as e: logger.warning(f"[CloudClient] steer failed: {e}") @@ -901,16 +1129,28 @@ class CloudClient(LinkAIClient): try: from agent.memory.conversation_store import get_conversation_store - store = get_conversation_store() + # Scope the lookup to the requested agent's workspace so each agent + # only sees its own history; None keeps the default-agent behaviour. + workspace = self._agent_workspace(payload.get("agent_id") or payload.get("agentId")) + store = get_conversation_store(workspace) result = store.load_history_page( session_id=session_id, page=page, page_size=page_size, ) + for turn in result.get("messages") or []: + extras = turn.get("extras") if isinstance(turn, dict) else None + if isinstance(extras, dict) and extras.get("agent_id"): + turn["agent_id"] = self._alias_agent_id(extras["agent_id"]) return { "action": "query", "payload": {"status": "success", **result}, } + except KeyError as e: + return { + "action": "query", + "payload": {"status": "error", "message": f"agent not found: {e.args[0] if e.args else ''}"}, + } except Exception as e: logger.error(f"[CloudClient] History query error: {e}") return { diff --git a/config.py b/config.py index 03af4f73..54bbfdb0 100644 --- a/config.py +++ b/config.py @@ -281,8 +281,11 @@ available_setting = { # as before. Each configured workspace is a complete CowAgent workspace. "agents": [], # Agent handling conversations that no channel instance binds. Defaults to - # the first configured agent when unset. + # the first configured agent when unset. When no agents list is configured, + # also sets the built-in single agent's id. "default_agent_id": "", + # Optional display name for the built-in single agent. + "default_agent_name": "", # Routes inbound conversations to an agent. Each entry needs channel_type # and agent_id; add conversation_id to bind one chat rather than the whole # channel. Unbound conversations go to default_agent_id.