Feature/multiaccount integrations

agent_core/core/impl/action/context.py

@@ -0,0 +1,41 @@
+"""Execution-scoped context for in-process actions.
+
+``current_input_data`` holds the full ``input_data`` dict of the action
+currently executing in this context. It exists so cross-cutting helpers
+deep inside an action's call tree (e.g. multi-account routing reading the
+``account`` hint) can see routing keys without threading them through
+every action function signature.
+
+Scope rules:
+  - Set only by the internal executors (``_atomic_action_internal*``),
+    reset in a ``finally`` — never leaks across actions.
+  - Sync actions run in a thread pool where the caller's context does NOT
+    propagate, so the executor wraps the call and sets the var inside the
+    worker thread (see ``run_with_input_context``).
+  - Sandboxed (subprocess) actions cannot see it at all — helpers must
+    treat a ``None`` value as "no context available".
+"""
+
+from __future__ import annotations
+
+from contextvars import ContextVar
+from typing import Any, Callable, Dict, Optional
+
+current_input_data: ContextVar[Optional[Dict[str, Any]]] = ContextVar(
+    "current_input_data", default=None
+)
+
+
+def run_with_input_context(
+    function_to_call: Callable[[dict], dict], input_data: dict
+) -> dict:
+    """Call a sync action with ``current_input_data`` set for its duration.
+
+    Used as the thread-pool target: the worker thread has its own context,
+    so the var must be set (and reset) inside the thread, not the caller.
+    """
+    token = current_input_data.set(input_data)
+    try:
+        return function_to_call(input_data)
+    finally:
+        current_input_data.reset(token)

agent_core/core/impl/action/executor.py

@@ -571,7 +571,9 @@ def _atomic_action_internal(
                 "The action_code string did not define a callable Python function."
             )
 
-        execution_result = function_to_call(input_data)
+        from agent_core.core.impl.action.context import run_with_input_context
+
+        execution_result = run_with_input_context(function_to_call, input_data)
         return execution_result
 
     except Exception as e:
@@ -618,16 +620,29 @@ async def _atomic_action_internal_async(
                 "The action_code string did not define a callable Python function."
             )
 
+        from agent_core.core.impl.action.context import (
+            current_input_data,
+            run_with_input_context,
+        )
+
         # Check if the function is async (coroutine function)
         if inspect.iscoroutinefunction(function_to_call):
             logger.debug(f"[ASYNC] Action '{action_name}' is async, awaiting directly")
-            execution_result = await function_to_call(input_data)
+            ctx_token = current_input_data.set(input_data)
+            try:
+                execution_result = await function_to_call(input_data)
+            finally:
+                current_input_data.reset(ctx_token)
         else:
-            # Sync function - run in thread pool to avoid blocking
+            # Sync function - run in thread pool to avoid blocking. The
+            # worker thread doesn't inherit this context, so the wrapper
+            # sets current_input_data inside the thread.
             logger.debug(
                 f"[SYNC] Action '{action_name}' is sync, running in thread pool"
             )
-            thread_future = THREAD_POOL.submit(function_to_call, input_data)
+            thread_future = THREAD_POOL.submit(
+                run_with_input_context, function_to_call, input_data
+            )
             try:
                 execution_result = await asyncio.wrap_future(thread_future)
             except asyncio.CancelledError:

agent_core/core/prompts/action.py

@@ -81,7 +81,10 @@
 
 Message Routing:
 - To reply to the user, send on the platform the incoming message came from —
-  check its source in the event stream.
+  check its source in the event stream. An event labeled just "user message"
+  (no platform tag) was typed in the local CraftBot interface: reply with
+  send_message, NOT a platform send action, even if earlier turns in this
+  session came from an external platform.
 - To act on a platform the user explicitly names, use that platform's send
   action (load its action set first if needed).
 - send_message and send_message_with_attachment ONLY records to the local 
@@ -106,6 +109,22 @@
   3. Read configuration of your own in app/config/.
 - Only ask the user if all three sources fail to provide the answer.
 
+Multi-Account Integrations:
+- Integrations can hold several connected accounts (e.g. a work and a school
+  Gmail). Every integration action takes an optional "account" input: an
+  email/identity, the user's nickname for the account, or any unique
+  fragment of either. Omitted = the primary account.
+- When the user names an account in ANY form ("my school calendar", "the
+  work inbox", "from my personal email"), extract that qualifier into
+  "account". Never silently default to primary when a qualifier is present.
+- If an account hint doesn't resolve, the action returns an error listing
+  the connected accounts — pick the right one from that list or ask the
+  user; do not retry the same hint.
+- IDs are account-scoped: a message/event/file id returned with
+  account="work" must be passed back with account="work" on follow-ups.
+- For irreversible actions (send, delete, clear) with multiple accounts
+  connected and no qualifier in the request: ask which account first.
+
 Critical Rules:
 - The selected action MUST be from the actions list. If none suitable, set
   action_name to "" (empty string).

app/agent_base.py

@@ -246,6 +246,19 @@ def __init__(
         self.db_interface = self._build_db_interface(
             data_dir=data_dir, chroma_path=chroma_path
         )
+        # Multi-account bridge: legacy actions of bridged platforms get the
+        # ``account`` input injected post-discovery (schemas are read live
+        # from the registry at prompt build, so this must run before the
+        # first turn). Never fatal — a failure just means those actions
+        # keep their pre-multi-account schemas this run.
+        try:
+            from app.data.action.integrations.account_bridge import (
+                inject_account_schemas,
+            )
+
+            inject_account_schemas()
+        except Exception as e:
+            logger.warning(f"[ACCOUNT_BRIDGE] schema injection failed: {e}")
 
         # LLM + prompt plumbing (may be deferred if API key not yet configured)
         self.llm = LLMInterface(
@@ -2272,12 +2285,20 @@ async def _handle_chat_message(self, payload: Dict):
                 trigger_payload["workflow_skills"] = payload["pre_selected_skills"]
 
             # Steer the action-selection LLM to use the right platform-specific
-            # send action when replying.
-            platform_hint = ""
+            # send action when replying. The UI case needs an explicit hint
+            # too: after a platform exchange in the same session, a bare
+            # message pattern-matches the previous "reply on <platform>"
+            # instruction and the reply leaks to that platform (observed
+            # live 2026-08-12: web-chat message answered on WhatsApp).
             if platform and platform.lower() != "craftbot interface":
                 platform_hint = (
                     f" from {platform} (reply on {platform}, NOT send_message)"
                 )
+            else:
+                platform_hint = (
+                    " typed in the CraftBot chat interface (reply with "
+                    "send_message, NOT a platform send action)"
+                )
             if is_third_party:
                 platform_hint += (
                     " — this is a third-party message; you may use the "
@@ -3357,11 +3378,38 @@ async def _initialize_external_libraries(self) -> None:
                 "openai_api_key": os.environ.get("OPENAI_API_KEY", ""),
             },
         )
+        # Every platform with a v2 provider (full port or auth-layer bridge)
+        # gets its listening from the ListenerManager's per-account fan-out;
+        # the legacy manager must not double-listen on any of them. Derived
+        # from the registry so newly bridged platforms are excluded
+        # automatically. Remaining legacy integrations keep legacy listening.
+        try:
+            from app.integrations import get_system
+
+            v2_platform_ids = [p.id for p in get_system().providers()]
+        except Exception as e:
+            logger.warning(
+                f"[EXT LIBS] v2 registry unavailable, falling back to static "
+                f"listener exclusions: {e}"
+            )
+            v2_platform_ids = ["gmail", "outlook", "slack"]
         self._external_comms = await initialize_manager(
-            on_message=self._handle_external_event
+            on_message=self._handle_external_event,
+            exclude_platforms=v2_platform_ids,
         )
         logger.info("[EXT LIBS] External integrations configured + manager started")
 
+        try:
+            from app.integrations import start_listeners
+
+            await start_listeners()
+            logger.info("[EXT LIBS] integrations listener manager started")
+        except Exception as e:
+            import traceback
+
+            logger.warning(f"[EXT LIBS] integrations listener manager failed to start: {e}")
+            logger.debug(f"[EXT LIBS] Traceback: {traceback.format_exc()}")
+
     # =====================================
     # Memory at startup
     # =====================================