refactor(scheduler): resolve live channel instance for delivery and use one global task store

Route scheduled delivery through the live ChannelManager instance (by instance_id, then the trusted recipient directory) instead of a bare create_channel() singleton, which fixed WeChat tasks stuck as "not ready" after restart in multi-agent mode (#3120). A shared common/channel_registry avoids the __main__-vs-app duplicate module hiding the manager from the scheduler.

Consolidate per-Agent tasks.json into a single global store; each task carries agent_id, and IM-task ownership is derived live from its delivery instance's binding (effective_task_agent_id) so re-binding a channel moves the task with no rewrite. Legacy per-Agent stores migrate once at startup; recipients.json moves under scheduler/.

Co-authored-by: cowagent <cow@cowagent.ai>

agent/tools/scheduler/integration.py

@@ -11,14 +11,38 @@ from common.utils import expand_path
 from bridge.context import Context, ContextType
 from bridge.reply import Reply, ReplyType
 
-# Backward-compatible aliases for the configured default agent.
+# One global scheduler for the whole process. Tasks are no longer sharded per
+# Agent workspace; the single store at ``scheduler_file_global()`` holds every
+# Agent's tasks, each tagged with its own ``agent_id`` (the Agent it runs as) and
+# ``instance_id`` (the channel login it delivers through). One service scans that
+# store; before running a task it scopes the runtime identity to the task's
+# ``agent_id``. This decouples task ownership from a channel instance's current
+# Agent binding: re-binding an instance moves no task data.
 _scheduler_service = None
 _task_store = None
-_scheduler_services: Dict[str, object] = {}
-_task_stores: Dict[str, object] = {}
-_recipient_stores: Dict[str, object] = {}
+# The recipient directory is shared across every Agent (state_dir
+# .scheduler_recipients_file), a single process-wide instance not one per workspace.
+_recipient_store = None
 # Module-level lock to guard idempotent initialization across threads
 _init_lock = threading.RLock()
+# One-shot fold of per-Agent ``scheduler/tasks.json`` files into the global store.
+_legacy_stores_migrated = False
+
+
+def _get_shared_recipient_store():
+    """The one recipient directory every Agent reads and writes.
+
+    Lives beside the global tasks.json at ``scheduler/recipients.json``; shared,
+    not per Agent, so any Agent's console can target anyone the instance has met.
+    """
+    global _recipient_store
+    if _recipient_store is None:
+        with _init_lock:
+            if _recipient_store is None:
+                from agent.tools.scheduler.recipient_store import RecipientStore
+                from common.state_dir import scheduler_recipients_file
+                _recipient_store = RecipientStore(str(scheduler_recipients_file()))
+    return _recipient_store
 
 
 def _resolve_workspace(workspace_root: str = None, agent_id: str = None):
@@ -37,77 +61,237 @@ def _resolve_workspace(workspace_root: str = None, agent_id: str = None):
     return str(state_root(identity))
 
 
+def _migrate_legacy_task_stores(task_store) -> None:
+    """Fold per-Agent ``scheduler/tasks.json`` files into the one global store.
+
+    The global file *is* the default Agent's historical path
+    (``shared_root/scheduler/tasks.json``), so those tasks stay put — we only
+    stamp a missing ``agent_id``. Every other Agent's workspace file is imported
+    once (skipping ids that already exist globally) and then renamed
+    ``.migrated`` so a later boot does not re-import them.
+
+    Re-binding a channel instance never calls this and never moves a task:
+    ownership lives on the task's ``agent_id`` field, not in the file path.
+    """
+    global _legacy_stores_migrated
+    if _legacy_stores_migrated:
+        return
+    try:
+        from agent.registry import get_agent_registry
+        from common.state_dir import scheduler_file, scheduler_file_global
+
+        registry = get_agent_registry()
+        default_id = registry.default_agent_id
+        global_path = os.path.realpath(str(scheduler_file_global()))
+
+        tasks = task_store.load_tasks()
+        changed = False
+
+        # Stamp missing owner on tasks already in the global file.
+        for task in tasks.values():
+            if not (task.get("agent_id") or "").strip():
+                task["agent_id"] = default_id
+                changed = True
+
+        for profile in registry.list(include_disabled=True):
+            legacy_path = os.path.realpath(str(scheduler_file(base=profile.workspace)))
+            if legacy_path == global_path:
+                continue
+            if not os.path.exists(legacy_path):
+                continue
+            try:
+                legacy_store = type(task_store)(legacy_path)
+                incoming = legacy_store.load_tasks()
+            except Exception as e:
+                logger.warning(
+                    f"[Scheduler] Skip legacy store {legacy_path}: {e}"
+                )
+                continue
+            imported = 0
+            skipped = 0
+            for task_id, task in incoming.items():
+                if task_id in tasks:
+                    skipped += 1
+                    continue
+                task = dict(task)
+                if not (task.get("agent_id") or "").strip():
+                    task["agent_id"] = profile.id
+                tasks[task_id] = task
+                imported += 1
+                changed = True
+            migrated_path = legacy_path + ".migrated"
+            try:
+                os.replace(legacy_path, migrated_path)
+            except OSError as e:
+                logger.warning(
+                    f"[Scheduler] Imported {imported} task(s) from {legacy_path} "
+                    f"but could not rename it aside: {e}"
+                )
+            else:
+                logger.info(
+                    f"[Scheduler] Migrated {imported} task(s) from {legacy_path} "
+                    f"into the global store"
+                    + (f" (skipped {skipped} duplicate id(s))" if skipped else "")
+                )
+
+        if changed:
+            task_store.save_tasks(tasks)
+        _legacy_stores_migrated = True
+    except Exception as e:
+        logger.warning(f"[Scheduler] Legacy task-store migration failed: {e}")
+
+
+def _resolve_instance_agent_id(instance_id: str) -> str:
+    """The Agent a channel instance is currently bound to, or "" for none.
+
+    Read live from the channel-instance config (team.json) rather than cached on
+    the task, so re-binding an instance to another Agent takes effect on the next
+    tick with no data migration. An unbound / legacy single-instance channel
+    (instance_id == channel_type, no explicit binding) returns "".
+    """
+    instance_id = (instance_id or "").strip()
+    if not instance_id:
+        return ""
+    try:
+        from channel.channel_instances import get_instance
+        from config import conf
+
+        inst = get_instance(conf(), instance_id)
+        if inst and (inst.agent_id or "").strip():
+            return inst.agent_id.strip()
+    except Exception as e:
+        logger.debug(
+            f"[Scheduler] Could not resolve agent for instance '{instance_id}': {e}"
+        )
+    return ""
+
+
+def effective_task_agent_id(task: dict, registry=None, validate: bool = False) -> str:
+    """Who a task belongs to *right now*.
+
+    This is the one place that decides ownership so execution and the console
+    never disagree. For an IM task the delivery instance's current binding wins
+    (re-binding a channel silently moves its tasks); otherwise the stored
+    ``agent_id`` is used; falling back to the default Agent.
+
+    ``validate`` gates whether an id must resolve to an *enabled* Agent:
+      * execution passes ``validate=True`` so a tick never runs as a removed or
+        disabled Agent (it degrades to the default instead);
+      * display callers (the list endpoint / ``list_tasks`` bucketing) leave it
+        False so a task keeps showing its stored owner even in contexts without
+        a live registry (e.g. unit tests) — the id is a label there, not an
+        identity to assume.
+
+    ``registry`` is optional; when None it is resolved lazily and, if that
+    fails, validation is skipped and the default falls back to "".
+    """
+    if registry is None:
+        try:
+            from agent.registry import get_agent_registry
+            registry = get_agent_registry()
+        except Exception:
+            registry = None
+
+    def _take(agent_id: str) -> str:
+        agent_id = (agent_id or "").strip()
+        if not agent_id:
+            return ""
+        if not validate or registry is None:
+            return agent_id
+        try:
+            return registry.get(agent_id, require_enabled=True).id
+        except Exception:
+            return ""
+
+    if isinstance(task, dict):
+        action = task.get("action") if isinstance(task.get("action"), dict) else {}
+        instance_id = (action.get("instance_id") or "").strip()
+        channel_type = (action.get("channel_type") or "").strip()
+        # Web targets a chat session, not a bound external instance, so it keeps
+        # using the stored agent_id; only external IM instances derive live.
+        if instance_id and channel_type not in ("", "web"):
+            derived = _take(_resolve_instance_agent_id(instance_id))
+            if derived:
+                return derived
+        stored = _take(task.get("agent_id"))
+        if stored:
+            return stored
+    if registry is not None:
+        return registry.default_agent_id
+    return ""
+
+
+def _resolve_task_agent_id(agent_bridge, task: dict) -> str:
+    """The Agent a task runs as, resolved against the live registry.
+
+    Thin wrapper over :func:`effective_task_agent_id` that validates (so a tick
+    never runs as a removed/disabled Agent) and always yields a concrete Agent
+    id (the default when nothing else resolves).
+    """
+    registry = agent_bridge.agent_registry
+    return (
+        effective_task_agent_id(task, registry, validate=True)
+        or registry.default_agent_id
+    )
+
+
 def init_scheduler(agent_bridge, workspace_root: str = None, agent_id: str = None) -> bool:
     """
-    Initialize scheduler service (idempotent).
+    Initialize the one global scheduler service (idempotent).
 
-    Safe to call multiple times and from multiple threads: only the first
-    successful call creates the singleton ``SchedulerService`` + background
-    scanning thread. Subsequent calls return immediately.
+    Safe to call any number of times and from any Agent: the scheduler is now a
+    single process-wide service over one global task store, so the first call
+    builds it and every later call (per-Agent warmup, roster reload, a lazy
+    first-message init) is a no-op that just returns True. The ``workspace_root``
+    / ``agent_id`` arguments are ignored — they only remain so existing per-Agent
+    call sites need no change.
 
     Args:
         agent_bridge: AgentBridge instance
 
     Returns:
-        True if scheduler is initialized (newly created or already running)
+        True if the scheduler is initialized (newly created or already running)
     """
     global _scheduler_service, _task_store
-    workspace_root = _resolve_workspace(workspace_root, agent_id)
-    agent_id = agent_bridge.agent_registry.get(agent_id).id
 
-    # Fast path: already initialized and running
-    service = _scheduler_services.get(workspace_root)
-    if service is not None and getattr(service, "running", False):
+    # Fast path: already running.
+    if _scheduler_service is not None and getattr(_scheduler_service, "running", False):
         return True
 
     with _init_lock:
-        # Re-check under the lock to avoid races where multiple threads
-        # passed the fast-path check before any of them acquired the lock.
-        service = _scheduler_services.get(workspace_root)
-        if service is not None and getattr(service, "running", False):
+        if _scheduler_service is not None and getattr(_scheduler_service, "running", False):
             return True
 
         try:
             from agent.tools.scheduler.task_store import TaskStore
-            from agent.tools.scheduler.recipient_store import RecipientStore
             from agent.tools.scheduler.scheduler_service import SchedulerService
+            from common.state_dir import scheduler_file_global
 
-            from common.state_dir import scheduler_file, scheduler_recipients_file
+            store_path = str(scheduler_file_global())
+            task_store = TaskStore(store_path)
+            _task_store = task_store
+            _migrate_legacy_task_stores(task_store)
+            logger.debug(f"[Scheduler] Global task store initialized: {store_path}")
 
-            store_path = str(scheduler_file(base=workspace_root))
+            _get_shared_recipient_store()
 
-            # Create task store (reuse if already created)
-            task_store = _task_stores.get(workspace_root)
-            if task_store is None:
-                task_store = TaskStore(store_path)
-                _task_stores[workspace_root] = task_store
-                logger.debug(f"[Scheduler] Task store initialized: {store_path}")
-
-            recipient_store = _recipient_stores.get(workspace_root)
-            if recipient_store is None:
-                recipient_store = RecipientStore(
-                    str(scheduler_recipients_file(base=workspace_root))
-                )
-                _recipient_stores[workspace_root] = recipient_store
-
-            # Create execute callback. Returns True on success, False to ask
-            # the scheduler to retry on the next tick (e.g. channel not yet
-            # ready right after process start).
+            # Execute callback. Returns True on success, False to retry next tick
+            # (e.g. channel not ready just after start). The Agent identity is
+            # taken from the task itself (not a closure), so one service runs
+            # tasks belonging to any Agent under the right workspace/memory/tools.
             def execute_task_callback(task: dict):
-                # Scheduler threads carry no identity of their own. Binding it
-                # here is the counterpart to chat_channel._handle: leaf code
-                # reached from a task (ToolManager, tmp_dir, memory) then lands
-                # in this Agent's workspace rather than the default one's.
                 from common.runtime_identity import identity_scope
 
+                agent_id = _resolve_task_agent_id(agent_bridge, task)
                 try:
                     with identity_scope(agent_id=agent_id):
                         action = task.get("action", {})
                         action_type = action.get("type")
                         channel_type = _primary_channel_type(action.get("channel_type"))
                         receiver = action.get("receiver", "")
+                        instance_id = action.get("instance_id") or ""
 
-                        if not _is_channel_ready(channel_type, receiver, agent_id):
+                        if not _is_channel_ready(channel_type, receiver, agent_id, instance_id):
                             logger.warning(
                                 f"[Scheduler] Task {task.get('id')}: channel "
                                 f"'{channel_type}' not ready for receiver={receiver} "
@@ -130,18 +314,11 @@ def init_scheduler(agent_bridge, workspace_root: str = None, agent_id: str = Non
                     logger.error(f"[Scheduler] Error executing task {task.get('id')}: {e}")
                     return False
 
-            # Create scheduler service
             service = SchedulerService(task_store, execute_task_callback)
             service.start()
-            _scheduler_services[workspace_root] = service
-            if agent_id == agent_bridge.agent_registry.default_agent_id:
-                _scheduler_service = service
-                _task_store = task_store
+            _scheduler_service = service
 
-            logger.info(
-                f"[Scheduler] Service initialized for agent={agent_id}, "
-                f"workspace={workspace_root}"
-            )
+            logger.info(f"[Scheduler] Global service initialized, store={store_path}")
             return True
 
         except Exception as e:
@@ -165,8 +342,150 @@ def _primary_channel_type(raw) -> str:
     return first or "unknown"
 
 
+def _is_live_channel_instance(channel) -> bool:
+    """True when *channel* is one ChannelManager is actually running.
+
+    Distinguishes a real, started instance (which reflects real login/token
+    state) from the bare singleton ``create_channel`` hands back when nothing is
+    running. Only for the former does an empty token map mean "not ready"; for
+    the latter it means "we can't tell", so callers must not hard-block on it.
+    """
+    try:
+        from common.channel_registry import get_channel_manager
+        manager = get_channel_manager()
+        if manager is None:
+            return False
+        for _name, ch in manager.find_channels_by_type(getattr(channel, "channel_type", "")):
+            if ch is channel:
+                return True
+    except Exception:
+        return False
+    return False
+
+
+def _channel_holds_receiver(channel, channel_type: str, receiver: str) -> bool:
+    """Whether *channel* already knows how to reach *receiver*.
+
+    Used to disambiguate when several running instances of one type exist and
+    the task carries no instance_id: the right one is the login that actually
+    holds this receiver's context token. Only weixin keeps such per-receiver
+    tokens; for other types we can't tell, so we don't claim a match.
+    """
+    if channel_type == "weixin":
+        tokens = getattr(channel, "_context_tokens", None)
+        return bool(tokens and receiver in tokens)
+    return False
+
+
+def _recover_instance_id_from_directory(channel_type: str, receiver: str) -> str:
+    """The instance that first saw *receiver*, per the recipient directory.
+
+    A receiver id is only meaningful within the instance that issued it — a
+    Feishu ``open_id`` is scoped to one app, so sending it through a different
+    Feishu instance fails with ``open_id cross app``. When a task carries no
+    (or a stale) instance_id, the recipient directory — keyed by
+    ``(instance_id, receiver)`` — is the authoritative place to recover the
+    right instance. Returns "" when unknown.
+    """
+    receiver = str(receiver or "").strip()
+    channel_type = str(channel_type or "").strip()
+    if not receiver:
+        return ""
+    try:
+        store = _get_shared_recipient_store()
+        for entry in store.list():
+            if (
+                entry.get("channel_type") == channel_type
+                and entry.get("receiver") == receiver
+            ):
+                iid = str(entry.get("instance_id") or "").strip()
+                if iid:
+                    return iid
+    except Exception as e:
+        logger.debug(f"[Scheduler] recipient directory lookup failed: {e}")
+    return ""
+
+
+def _resolve_delivery_channel(channel_type: str, instance_id: str = "", receiver: str = ""):
+    """The channel object a scheduled task should send through.
+
+    Prefers the *live running instance* held by ChannelManager, looked up by
+    instance_id. That matters because a running channel carries state a freshly
+    built one does not: in-memory login/context tokens, an open websocket, the
+    per-instance credentials it started with. Building a bare
+    ``create_channel(channel_type)`` instead would grab whichever instance the
+    singleton cache happens to hold — a never-started empty object once the real
+    bot runs as a channel_instance, which is exactly the delivery bug this avoids.
+
+    Resolution order:
+      1. Exact match by instance_id (or legacy instance_id == channel_type).
+      2. Task carries no (or a stale) instance_id — recover the right live
+         instance without ever sending to the wrong login:
+           a. Ask the recipient directory which instance first saw this
+              receiver, then use that instance. This is what keeps an
+              app-scoped id (a Feishu open_id) from being sent through a
+              different app instance (``open_id cross app``).
+           b. Else pick the instance already holding the receiver's token
+              (weixin keeps per-receiver tokens).
+           c. Else, only when exactly one instance of the type runs, use it.
+              With several running instances and no way to tell them apart we
+              refuse to guess (returning that lone-or-none live instance),
+              since a wrong pick delivers to the wrong account.
+      3. Fall back to ``create_channel`` (legacy single-instance) only when no
+         manager or no running instance of the type exists.
+    """
+    instance_id = str(instance_id or "").strip()
+    receiver = str(receiver or "").strip()
+    try:
+        from common.channel_registry import get_channel_manager
+        manager = get_channel_manager()
+        if manager is not None:
+            # (1) Exact key: instance_id, or channel_type for legacy singletons.
+            lookup = instance_id or channel_type
+            live = manager.get_channel(lookup)
+            if live is not None:
+                return live
+
+            candidates = manager.find_channels_by_type(channel_type)
+            if candidates:
+                # (2a) Authoritative: the directory knows which instance owns
+                # this receiver. Resolve to that exact instance.
+                recovered = _recover_instance_id_from_directory(channel_type, receiver)
+                if recovered:
+                    live = manager.get_channel(recovered)
+                    if live is not None:
+                        logger.info(
+                            f"[Scheduler] Recovered instance '{recovered}' for "
+                            f"{channel_type} receiver={receiver} from recipient directory"
+                        )
+                        return live
+
+                # (2b) Token-holder (weixin): the login that already knows them.
+                if receiver:
+                    for _name, ch in candidates:
+                        if _channel_holds_receiver(ch, channel_type, receiver):
+                            return ch
+
+                # (2c) Only safe to auto-pick when the type runs a single
+                # instance. Several instances with app-scoped receiver ids
+                # (Feishu open_id) must not be guessed — that would deliver to
+                # the wrong app. Leave it to the bare fallback / hard failure.
+                if len(candidates) == 1:
+                    return candidates[0][1]
+                logger.warning(
+                    f"[Scheduler] {channel_type} has {len(candidates)} running "
+                    f"instances but the task named none and receiver={receiver} "
+                    f"is not in the recipient directory; cannot pick safely"
+                )
+    except Exception as e:
+        logger.debug(f"[Scheduler] No live channel for instance '{instance_id}': {e}")
+
+    from channel.channel_factory import create_channel
+    return create_channel(channel_type)
+
+
 def _is_channel_ready(
-    channel_type: str, receiver: str, agent_id: str = None
+    channel_type: str, receiver: str, agent_id: str = None, instance_id: str = ""
 ) -> bool:
     """Best-effort readiness probe for outbound channels.
 
@@ -174,18 +493,31 @@ def _is_channel_ready(
     logged in, web session has no polling queue), so the scheduler can
     defer instead of consuming the task. Unknown channels return True
     to preserve previous behaviour.
+
+    Probes the same live instance delivery will use, so a "ready" answer means
+    *that* instance is ready, not some other login of the same channel type.
     """
     if not channel_type or channel_type == "unknown":
         return True
     try:
-        from channel.channel_factory import create_channel
-        channel = create_channel(channel_type)
+        channel = _resolve_delivery_channel(channel_type, instance_id, receiver)
         if channel is None:
             return False
 
         if channel_type == "weixin":
             tokens = getattr(channel, "_context_tokens", None)
             if not tokens or receiver not in tokens:
+                # The token check only means something for the *running* login.
+                # If we're staring at a bare, never-started singleton (no live
+                # instance in the manager), its token map is empty no matter
+                # what — blocking here would defer forever (issue #3120). Fall
+                # back to the pre-check behaviour and let the send attempt run.
+                if not _is_live_channel_instance(channel):
+                    logger.debug(
+                        "[Scheduler] weixin readiness inconclusive (no live "
+                        f"instance for receiver={receiver}); allowing send attempt"
+                    )
+                    return True
                 return False
             return True
 
@@ -204,66 +536,66 @@ def _is_channel_ready(
 
 
 def get_task_store(workspace_root: str = None, agent_id: str = None):
-    """Get the task store owned by one agent workspace."""
-    workspace_root = _resolve_workspace(workspace_root, agent_id)
-    return _task_stores.get(workspace_root)
+    """The one global task store. ``workspace_root``/``agent_id`` are ignored
+    (kept for call-site compatibility); tasks are filtered by ``agent_id`` at
+    the query, not by which store they live in."""
+    if _task_store is not None:
+        return _task_store
+    # Lazily build the store even if the service hasn't started yet, so tools /
+    # handlers that only read or write tasks work before scheduler warmup.
+    with _init_lock:
+        if _task_store is not None:
+            return _task_store
+        try:
+            from agent.tools.scheduler.task_store import TaskStore
+            from common.state_dir import scheduler_file_global
+
+            globals()["_task_store"] = TaskStore(str(scheduler_file_global()))
+            _migrate_legacy_task_stores(globals()["_task_store"])
+        except Exception as e:
+            logger.error(f"[Scheduler] Failed to open global task store: {e}")
+            return None
+    return _task_store
 
 
 def get_scheduler_service(workspace_root: str = None, agent_id: str = None):
-    """Get the scheduler service owned by one agent workspace."""
-    workspace_root = _resolve_workspace(workspace_root, agent_id)
-    return _scheduler_services.get(workspace_root)
+    """The one global scheduler service (``workspace_root``/``agent_id`` ignored)."""
+    return _scheduler_service
 
 
 def get_recipient_store(workspace_root: str = None, agent_id: str = None):
-    """Get the trusted recipient directory owned by one agent workspace."""
-    workspace_root = _resolve_workspace(workspace_root, agent_id)
-    return _recipient_stores.get(workspace_root)
+    """Get the shared trusted recipient directory.
+
+    Shared across Agents, so the ``workspace_root``/``agent_id`` arguments are
+    ignored; they are kept only so existing call sites need no change.
+    """
+    return _get_shared_recipient_store()
 
 
 def reset_scheduler_services(stop: bool = True) -> None:
-    """Stop and forget all scheduler services, primarily for reloads/tests."""
-    global _scheduler_service, _task_store
+    """Stop and forget the global scheduler service, primarily for reloads/tests."""
+    global _scheduler_service, _task_store, _recipient_store, _legacy_stores_migrated
     with _init_lock:
-        if stop:
-            for service in list(_scheduler_services.values()):
-                try:
-                    service.stop()
-                except Exception:
-                    pass
-        _scheduler_services.clear()
-        _task_stores.clear()
-        _recipient_stores.clear()
+        if stop and _scheduler_service is not None:
+            try:
+                _scheduler_service.stop()
+            except Exception:
+                pass
+        _recipient_store = None
         _scheduler_service = None
         _task_store = None
+        _legacy_stores_migrated = False
 
 
 def stop_scheduler(agent_id: str = None, workspace_root: str = None) -> bool:
-    """Stop and forget a single Agent's scheduler service, leaving the rest of
-    the fleet running. Used when an Agent is archived/removed so a roster edit
-    does not have to reset every scheduler. Returns True if one was found.
+    """No-op under the global scheduler.
 
-    The actual stop is detached to a daemon thread: ``service.stop()`` joins the
-    scan loop (up to a few seconds), and a roster edit should not block the HTTP
-    response on that. We drop the service from the registry synchronously so it
-    is immediately forgotten; the thread just winds the loop down."""
-    workspace_root = _resolve_workspace(workspace_root, agent_id)
-    with _init_lock:
-        service = _scheduler_services.pop(workspace_root, None)
-        _task_stores.pop(workspace_root, None)
-    if service is None:
-        return False
-    threading.Thread(
-        target=lambda: _safe_stop(service), daemon=True, name="scheduler-stop"
-    ).start()
-    return True
-
-
-def _safe_stop(service) -> None:
-    try:
-        service.stop()
-    except Exception:
-        pass
+    Tasks are no longer sharded per Agent, so removing one Agent from the roster
+    does not stop any scheduler — the single global service keeps running the
+    remaining Agents' tasks. A removed/disabled Agent's tasks simply stop
+    matching at execution time (they fall back to the default owner or are
+    skipped). Kept for call-site compatibility; always returns False."""
+    return False
 
 
 def _remember_delivered_output(
@@ -406,10 +738,9 @@ def _execute_agent_task(task: dict, agent_bridge, agent_id: str = None) -> bool:
                 )
                 return True
 
-            from channel.channel_factory import create_channel
-            channel = create_channel(channel_type)
+            channel = _resolve_delivery_channel(channel_type, action.get("instance_id") or "", receiver)
             if not channel:
-                logger.error(f"[Scheduler] Failed to create channel: {channel_type}")
+                logger.error(f"[Scheduler] Failed to resolve channel: {channel_type}")
                 return False
 
             if channel_type == "web" and hasattr(channel, 'request_to_session'):
@@ -495,13 +826,11 @@ def _execute_send_message(task: dict, agent_bridge, agent_id: str = None) -> boo
 
         # Create reply
         reply = Reply(ReplyType.TEXT, content)
-        
-        # Get channel and send
-        from channel.channel_factory import create_channel
-        
-        channel = create_channel(channel_type)
+
+        # Get the live instance for this task and send through it.
+        channel = _resolve_delivery_channel(channel_type, action.get("instance_id") or "", receiver)
         if not channel:
-            logger.error(f"[Scheduler] Failed to create channel: {channel_type}")
+            logger.error(f"[Scheduler] Failed to resolve channel: {channel_type}")
             return False
 
         if channel_type == "web" and hasattr(channel, 'request_to_session'):
@@ -575,10 +904,9 @@ def _execute_tool_call(task: dict, agent_bridge, agent_id: str = None) -> bool:
 
         reply = Reply(ReplyType.TEXT, content)
 
-        from channel.channel_factory import create_channel
-        channel = create_channel(channel_type)
+        channel = _resolve_delivery_channel(channel_type, action.get("instance_id") or "", receiver)
         if not channel:
-            logger.error(f"[Scheduler] Failed to create channel: {channel_type}")
+            logger.error(f"[Scheduler] Failed to resolve channel: {channel_type}")
             return False
 
         if channel_type == "web" and request_id and hasattr(channel, 'request_to_session'):
@@ -660,10 +988,9 @@ def _execute_skill_call(task: dict, agent_bridge, agent_id: str = None) -> bool:
         if result_prefix:
             content = f"{result_prefix}\n\n{content}"
 
-        from channel.channel_factory import create_channel
-        channel = create_channel(channel_type)
+        channel = _resolve_delivery_channel(channel_type, action.get("instance_id") or "", receiver)
         if not channel:
-            logger.error(f"[Scheduler] Failed to create channel: {channel_type}")
+            logger.error(f"[Scheduler] Failed to resolve channel: {channel_type}")
             return False
 
         if channel_type == "web" and hasattr(channel, 'request_to_session'):
@@ -720,10 +1047,15 @@ def attach_scheduler_to_tool(tool, context: Context = None):
         # persist ephemeral credentials; the channel keeps enforcing its own
         # authentication and readiness rules at delivery time.
         if recipient_store and channel_type != "web":
+            # instance_id is what a scheduled delivery must route back through:
+            # two instances of one channel type have separate logins and receiver
+            # id spaces. It rides on the inbound context; when absent (legacy
+            # single-instance channel) the store falls back to the channel type.
             recipient_store.remember(
                 channel_type,
                 context.get("receiver"),
                 name=tool._get_receiver_name(context),
                 is_group=context.get("isgroup", False),
                 session_id=context.get("session_id"),
+                instance_id=context.get("instance_id") or "",
             )

agent/tools/scheduler/recipient_store.py

@@ -16,6 +16,21 @@ class RecipientStore:
     The store deliberately contains no access tokens or channel credentials.
     Channel implementations remain responsible for authentication and their
     normal outbound readiness checks.
+
+    The directory is shared across Agents, but keyed by ``(instance_id, receiver)``
+    rather than by ``(channel_type, receiver)``. A channel *type* can run several
+    *instances* (two Feishu bots, two WeChat logins), each with its own login,
+    tokens and — crucially — its own id space for receivers: the same receiver
+    string on two WeChat instances is two different people. Delivery therefore has
+    to go back out through the exact instance that first saw the contact, so the
+    instance is part of the identity, not just the type.
+
+    Legacy single-instance installs use ``instance_id == channel_type`` (that is
+    how the launcher already synthesizes them), so an old entry saved under just a
+    channel type keeps resolving unchanged.
+
+    The owning Agent of a scheduled delivery is derived from the recipient's
+    channel instance (an instance binds to one Agent), so it is not stored here.
     """
 
     def __init__(self, store_path: str) -> None:
@@ -24,8 +39,12 @@ class RecipientStore:
         self.store_path.parent.mkdir(parents=True, exist_ok=True)
 
     @staticmethod
-    def _key(channel_type: str, receiver: str) -> str:
-        return f"{channel_type}\0{receiver}"
+    def _key(instance_id: str, receiver: str) -> str:
+        # ``:`` reads cleanly in the on-disk JSON. Both an instance_id and a
+        # receiver id can themselves carry a colon (feishu group ids do), so the
+        # key is only ever a joined form; downstream code uses the structured
+        # fields, never a split of this key.
+        return f"{instance_id}:{receiver}"
 
     def _load_unlocked(self) -> Dict[str, dict]:
         if not self.store_path.exists():
@@ -47,6 +66,19 @@ class RecipientStore:
             os.fsync(handle.fileno())
         os.replace(temporary, self.store_path)
 
+    @staticmethod
+    def _normalize(entry: dict) -> dict:
+        """Fill fields an older on-disk entry may predate, without rewriting it.
+
+        ``instance_id`` predates multi-instance and defaults to the channel type,
+        which is exactly the legacy single-instance id, so an old entry resolves
+        and delivers the same as before.
+        """
+        result = dict(entry)
+        if not result.get("instance_id"):
+            result["instance_id"] = result.get("channel_type", "")
+        return result
+
     def remember(
         self,
         channel_type: str,
@@ -55,13 +87,18 @@ class RecipientStore:
         name: str = "",
         is_group: bool = False,
         session_id: str = "",
+        instance_id: str = "",
     ) -> Optional[dict]:
         channel_type = str(channel_type or "").strip()
         receiver = str(receiver or "").strip()
+        # A missing instance_id means a legacy single-instance channel, whose id
+        # is the channel type; this keeps old callers and old data on one path.
+        instance_id = str(instance_id or "").strip() or channel_type
         if not channel_type or not receiver or channel_type in {"unknown", "web"}:
             return None
         entry = {
             "channel_type": channel_type,
+            "instance_id": instance_id,
             "receiver": receiver,
             "name": str(name or receiver),
             "is_group": bool(is_group),
@@ -70,26 +107,44 @@ class RecipientStore:
         }
         with self._lock:
             recipients = self._load_unlocked()
-            key = self._key(channel_type, receiver)
+            key = self._key(instance_id, receiver)
             previous = recipients.get(key)
-            stable_fields = ("channel_type", "receiver", "name", "is_group", "session_id")
-            if previous and all(previous.get(field) == entry[field] for field in stable_fields):
-                try:
-                    last_seen = datetime.fromisoformat(previous["last_seen_at"])
-                    if datetime.now(timezone.utc) - last_seen < timedelta(hours=1):
-                        return dict(previous)
-                except (KeyError, TypeError, ValueError):
-                    pass
+            if previous:
+                stable_fields = (
+                    "channel_type",
+                    "instance_id",
+                    "receiver",
+                    "name",
+                    "is_group",
+                    "session_id",
+                )
+                unchanged = all(previous.get(field) == entry[field] for field in stable_fields)
+                if unchanged:
+                    try:
+                        last_seen = datetime.fromisoformat(previous["last_seen_at"])
+                        if datetime.now(timezone.utc) - last_seen < timedelta(hours=1):
+                            return dict(previous)
+                    except (KeyError, TypeError, ValueError):
+                        pass
             recipients[key] = entry
             self._save_unlocked(recipients)
         return dict(entry)
 
-    def get(self, channel_type: str, receiver: str) -> Optional[dict]:
+    def get(self, instance_id: str, receiver: str) -> Optional[dict]:
+        """Resolve one recipient by the instance that saw them.
+
+        ``instance_id`` may be a legacy channel type, which is the id a
+        single-instance channel runs under, so old lookups keep working.
+        """
+        instance_id = str(instance_id or "").strip()
         with self._lock:
-            entry = self._load_unlocked().get(self._key(channel_type, receiver))
-        return dict(entry) if entry else None
+            entry = self._load_unlocked().get(self._key(instance_id, receiver))
+        return self._normalize(entry) if entry else None
 
     def list(self) -> List[dict]:
         with self._lock:
-            entries = [dict(item) for item in self._load_unlocked().values()]
-        return sorted(entries, key=lambda item: (item["channel_type"], item["name"], item["receiver"]))
+            entries = [self._normalize(item) for item in self._load_unlocked().values()]
+        return sorted(
+            entries,
+            key=lambda item: (item["channel_type"], item["name"], item["receiver"]),
+        )

agent/tools/scheduler/scheduler_tool.py

@@ -32,7 +32,7 @@ class SchedulerTool(BaseTool):
         "- cron: cron表达式，如'0 8 * * *'表示每天8点\n\n"
         "注意：'X秒后'用once+相对时间，'每X秒'用interval\n"
         "For Web cross-channel delivery, call action='list_recipients' first, "
-        "then create a fixed-message task using the returned channel_type and receiver."
+        "then create a task (fixed message or ai_task) using the returned channel_type and receiver."
     )
     params: dict = {
         "type": "object",
@@ -77,7 +77,11 @@ class SchedulerTool(BaseTool):
             },
             "channel_type": {
                 "type": "string",
-                "description": "Target channel for a Web-created fixed-message task. Use list_recipients first."
+                "description": "Target channel type for a Web-created cross-channel task. Use list_recipients first."
+            },
+            "instance_id": {
+                "type": "string",
+                "description": "The specific channel instance the recipient belongs to, as returned by list_recipients. Required when a channel type runs more than one instance; the same receiver on two instances is two different targets. Omit for a single-instance channel."
             },
             "receiver": {
                 "type": "string",
@@ -179,17 +183,19 @@ class SchedulerTool(BaseTool):
 
         target_channel = kwargs.get("channel_type")
         target_receiver = kwargs.get("receiver")
+        # A channel type can run several instances; delivery must go back through
+        # the exact one that saw the contact. When the caller does not name an
+        # instance (legacy single-instance channel), it equals the channel type.
+        target_instance = kwargs.get("instance_id") or target_channel
         target = None
         if target_channel or target_receiver:
             if context.get("channel_type") != "web":
                 return "Error: cross-channel recipients can only be selected from the Web console"
             if not target_channel or not target_receiver:
                 return "Error: channel_type and receiver must be provided together"
-            if ai_task:
-                return "Error: cross-channel delivery currently supports fixed messages only"
             if not self.recipient_store:
                 return "Error: trusted recipient directory is unavailable"
-            target = self.recipient_store.get(target_channel, target_receiver)
+            target = self.recipient_store.get(target_instance, target_receiver)
             if not target:
                 return "Error: target is not in the trusted recipient directory; use list_recipients"
         
@@ -204,6 +210,12 @@ class SchedulerTool(BaseTool):
         receiver_name = target["name"] if target else self._get_receiver_name(context)
         is_group = target["is_group"] if target else context.get("isgroup", False)
         channel_type = target["channel_type"] if target else self.config.get("channel_type", "unknown")
+        # The instance to deliver back through: the target's own for a
+        # cross-channel task, else this conversation's. Defaults to the channel
+        # type so a legacy single-instance channel keeps resolving as before.
+        instance_id = (
+            target.get("instance_id") if target else context.get("instance_id")
+        ) or channel_type
 
         # Build action based on message or ai_task
         if message:
@@ -214,6 +226,7 @@ class SchedulerTool(BaseTool):
                 "receiver_name": receiver_name,
                 "is_group": is_group,
                 "channel_type": channel_type,
+                "instance_id": instance_id,
                 "notify_session_id": notify_session_id,
             }
         else:  # ai_task
@@ -224,6 +237,7 @@ class SchedulerTool(BaseTool):
                 "receiver_name": receiver_name,
                 "is_group": is_group,
                 "channel_type": channel_type,
+                "instance_id": instance_id,
                 "notify_session_id": notify_session_id,
             }
             # silent only applies to ai_task; fixed messages always deliver
@@ -235,9 +249,15 @@ class SchedulerTool(BaseTool):
         if msg and hasattr(msg, 'sender_staff_id') and not context.get("isgroup", False):
             action["dingtalk_sender_staff_id"] = msg.sender_staff_id
         
+        # The Agent this task runs as. Stored on the task (not implied by which
+        # file it lives in) so the global scheduler can scope execution to it and
+        # so re-binding a channel instance to a different Agent never moves data.
+        owner_agent_id = self._resolve_owner_agent_id(context, target)
+
         task_data = {
             "id": task_id,
             "name": name,
+            "agent_id": owner_agent_id,
             "enabled": True,
             "created_at": datetime.now().isoformat(),
             "updated_at": datetime.now().isoformat(),
@@ -309,9 +329,10 @@ class SchedulerTool(BaseTool):
         lines = ["Trusted scheduler recipients:"]
         for item in recipients:
             kind = "group" if item["is_group"] else "user"
+            instance_id = item.get("instance_id") or item["channel_type"]
             lines.append(
                 f"- {item['name']} ({kind}): channel_type={item['channel_type']}, "
-                f"receiver={item['receiver']}"
+                f"instance_id={instance_id}, receiver={item['receiver']}"
             )
         return "\n".join(lines)
     
@@ -517,3 +538,43 @@ class SchedulerTool(BaseTool):
         except Exception:
             pass
         return "未知"
+
+    def _resolve_owner_agent_id(self, context: Context, target: Optional[dict]) -> str:
+        """Which Agent the new task runs as.
+
+        For a cross-channel task the owner is the Agent the target's channel
+        instance is bound to (delivery and identity should match how an inbound
+        message from that instance would route). For a normal conversation task
+        it is simply the Agent handling this conversation. Falls back to the
+        ambient runtime identity, then the default Agent.
+        """
+        # Cross-channel: derive from the target instance's binding.
+        if target:
+            iid = (target.get("instance_id") or "").strip()
+            if iid:
+                try:
+                    from channel.channel_instances import get_instance
+                    from config import conf as _conf
+
+                    inst = get_instance(_conf(), iid)
+                    if inst and getattr(inst, "agent_id", ""):
+                        return inst.agent_id
+                except Exception:
+                    pass
+        # Conversation task: the Agent on the context, else ambient identity.
+        agent_id = (context.get("agent_id") if context else "") or ""
+        if not agent_id:
+            try:
+                from common.runtime_identity import current_identity
+
+                agent_id = current_identity().agent_id or ""
+            except Exception:
+                agent_id = ""
+        if agent_id:
+            return agent_id
+        try:
+            from agent.registry import get_agent_registry
+
+            return get_agent_registry().default_agent_id
+        except Exception:
+            return ""

agent/tools/scheduler/task_store.py

@@ -178,21 +178,40 @@ class TaskStore:
         tasks = self.load_tasks()
         return tasks.get(task_id)
     
-    def list_tasks(self, enabled_only: bool = False) -> List[dict]:
+    def list_tasks(self, enabled_only: bool = False, agent_id: str = None) -> List[dict]:
         """
         List all tasks
-        
+
         Args:
             enabled_only: If True, only return enabled tasks
-            
+            agent_id: If given, only return tasks owned by this Agent. Ownership
+                is the task's *effective* owner: for an IM task that is the
+                delivery instance's current binding (so re-binding a channel
+                re-buckets its tasks with no data change), else the stored
+                ``agent_id``, else the default Agent. This keeps the per-Agent
+                list identical to what actually runs.
+
         Returns:
             List of task dictionaries
         """
         tasks = self.load_tasks()
         task_list = list(tasks.values())
-        
+
         if enabled_only:
             task_list = [t for t in task_list if t.get("enabled", True)]
+
+        if agent_id:
+            from agent.tools.scheduler.integration import effective_task_agent_id
+            default_id = ""
+            try:
+                from agent.registry import get_agent_registry
+                default_id = get_agent_registry().default_agent_id
+            except Exception:
+                pass
+            task_list = [
+                t for t in task_list
+                if (effective_task_agent_id(t) or default_id) == agent_id
+            ]
         
         # Sort by enabled status (enabled first), then by next_run_at
         def sort_key(t):

app.py

@@ -15,7 +15,12 @@ from plugins import *
 import threading
 
 
-_channel_mgr = None
+# The manager lives in common.channel_registry, not here: app.py runs as
+# ``__main__`` (``python app.py``), so a module-level global here would be
+# invisible to any code that does ``from app import ...`` (that import builds a
+# second, separate ``app`` module). The registry is a single shared cell every
+# caller sees — see common/channel_registry.py and issue #3120.
+from common.channel_registry import get_channel_manager, set_channel_manager
 
 # Desktop mode: a lighter runtime for the packaged Electron client. Plugins are
 # loaded in a background thread (so command plugins like cow_cli/godcmd work
@@ -23,10 +28,6 @@ _channel_mgr = None
 DESKTOP_MODE = os.environ.get("COW_DESKTOP") == "1"
 
 
-def get_channel_manager():
-    return _channel_mgr
-
-
 def _parse_channel_type(raw) -> list:
     """
     Parse channel_type config value into a list of channel names.
@@ -149,6 +150,27 @@ class ChannelManager:
     def get_channel(self, channel_name: str):
         return self._channels.get(channel_name)
 
+    def find_channels_by_type(self, channel_type: str) -> list:
+        """Every running channel whose type matches, keyed by registry name.
+
+        Instances register under their ``instance_id`` (e.g. ``weixin-c696...``),
+        so a caller that only knows the bare ``channel_type`` (an old scheduled
+        task whose action never stored an instance_id) can't hit ``get_channel``
+        directly. This lets it recover the live instance(s) of that type instead
+        of falling back to a never-started bare singleton. Returns ``[(name,
+        channel), ...]`` in registry order.
+        """
+        ctype = (channel_type or "").strip()
+        with self._lock:
+            items = list(self._channels.items())
+        out = []
+        for name, ch in items:
+            if ch is None:
+                continue
+            if getattr(ch, "channel_type", "") == ctype:
+                out.append((name, ch))
+        return out
+
     @staticmethod
     def _normalize_entry(entry):
         """Accept both a legacy channel-type string and a ChannelInstance.
@@ -705,7 +727,6 @@ def _scaffold_subagent_assets():
 
 
 def run():
-    global _channel_mgr
     try:
         # Before any TLS connection: a packaged build has no OpenSSL CA store.
         bundle = ensure_ca_bundle()
@@ -759,8 +780,9 @@ def run():
 
         logger.info(f"[App] Starting channels: {channel_names}")
 
-        _channel_mgr = ChannelManager()
-        _channel_mgr.start(channel_names, first_start=True)
+        channel_mgr = ChannelManager()
+        set_channel_manager(channel_mgr)
+        channel_mgr.start(channel_names, first_start=True)
 
         while True:
             time.sleep(1)

channel/channel_instances.py

@@ -92,6 +92,20 @@ CREDENTIAL_KEYS: Dict[str, tuple] = {
     ),
 }
 
+# Short human labels used only to seed a new instance's default name (e.g.
+# "微信 2"). Purely cosmetic and easily overridden by the user, so an unknown
+# type simply falls back to its raw channel_type string.
+_CHANNEL_TYPE_LABELS: Dict[str, str] = {
+    const.FEISHU: "飞书",
+    const.DINGTALK: "钉钉",
+    const.WECOM_BOT: "企微机器人",
+    const.WEIXIN: "微信",
+    const.QQ: "QQ",
+    const.TELEGRAM: "Telegram",
+    const.SLACK: "Slack",
+    const.DISCORD: "Discord",
+}
+
 # Channel types that actually support running more than one instance today.
 # Others may appear in channel_instances but will run as a single instance
 # (their @singleton is not yet bypassed); we log and fall back gracefully.
@@ -114,6 +128,10 @@ class ChannelInstance:
     instance_id: str
     channel_type: str
     agent_id: str = ""
+    #: Human-friendly label the user can edit (e.g. "运营微信"). Optional: when
+    #: empty the UI falls back to a credential-derived name or the instance id, so
+    #: an install that never sets one is unaffected.
+    name: str = ""
     credentials: Dict[str, Any] = field(default_factory=dict)
     #: Other Agents sharing this channel's conversations. ``agent_id`` is the
     #: owner/leader that receives every inbound message; ``members`` are the
@@ -175,6 +193,7 @@ def _explicit_instances(raw_list: list) -> List[ChannelInstance]:
         seen_ids.add(instance_id)
 
         agent_id = str(raw.get("agent_id") or "").strip()
+        name = str(raw.get("name") or "").strip()
 
         raw_members = raw.get("members")
         members: List[str] = []
@@ -206,6 +225,7 @@ def _explicit_instances(raw_list: list) -> List[ChannelInstance]:
                 instance_id=instance_id,
                 channel_type=channel_type,
                 agent_id=agent_id,
+                name=name,
                 credentials=creds,
                 members=members,
                 legacy=False,
@@ -295,6 +315,7 @@ def bootstrap_legacy_instances(
                 "instance_id": ctype,
                 "channel_type": ctype,
                 "agent_id": default_id,
+                "name": _CHANNEL_TYPE_LABELS.get(ctype, ctype),
                 "credentials": creds,
             }
         )
@@ -360,6 +381,23 @@ def _clean_members(members, owner_id: str) -> list:
     return out
 
 
+def default_instance_name(channel_type: str, records: Iterable[Mapping[str, Any]]) -> str:
+    """A friendly default label for a freshly created instance of *channel_type*.
+
+    Shape is "<type label>" for the first instance of a type and "<type label> N"
+    for later ones, so a second WeChat bot reads as "微信 2" rather than a random
+    id. Only used to seed the field; the user can rename it afterwards.
+    """
+    ctype = _normalize_type((channel_type or "").strip())
+    label = _CHANNEL_TYPE_LABELS.get(ctype, ctype)
+    same_type = sum(
+        1
+        for r in records
+        if _normalize_type(str(r.get("channel_type") or "").strip()) == ctype
+    )
+    return label if same_type == 0 else f"{label} {same_type + 1}"
+
+
 def upsert_instance(
     settings: Mapping[str, Any],
     channel_type: str,
@@ -367,14 +405,17 @@ def upsert_instance(
     agent_id: Optional[str] = None,
     credentials: Optional[Mapping[str, Any]] = None,
     members: Optional[list] = None,
+    name: Optional[str] = None,
 ) -> ChannelInstance:
     """Create or update one channel instance record and persist it.
 
     Matching is by ``instance_id``. When it is empty a new stable id is
-    generated. ``agent_id``, ``credentials`` and ``members`` are merged onto any
-    existing record so a partial update (e.g. credentials only) does not drop
-    the binding or the team. Pass ``members=None`` to leave the team untouched,
-    or ``members=[]`` to clear it. Returns the resulting :class:`ChannelInstance`.
+    generated. ``agent_id``, ``credentials``, ``members`` and ``name`` are merged
+    onto any existing record so a partial update (e.g. credentials only) does not
+    drop the binding, the team or the label. Pass ``members=None`` to leave the
+    team untouched, or ``members=[]`` to clear it. Pass ``name=None`` to leave the
+    label untouched; a brand-new record with no name is seeded with a friendly
+    default. Returns the resulting :class:`ChannelInstance`.
     """
     from agent import team
 
@@ -395,6 +436,12 @@ def upsert_instance(
         record["channel_type"] = ctype
         if agent_id is not None:
             record["agent_id"] = str(agent_id).strip()
+        if name is not None:
+            new_name = str(name).strip()
+            if new_name:
+                record["name"] = new_name
+            else:
+                record.pop("name", None)
         if incoming_creds:
             merged = dict(record.get("credentials") or {})
             merged.update(incoming_creds)
@@ -410,10 +457,17 @@ def upsert_instance(
         break
 
     if not updated:
+        # Seed a friendly default label for a new instance so the console has
+        # something readable to show before the user renames it. Computed against
+        # the records *before* this one is appended, so the first is unnumbered.
+        seeded_name = str(name).strip() if name is not None else ""
+        if not seeded_name:
+            seeded_name = default_instance_name(ctype, records)
         result_record = {
             "instance_id": target_id,
             "channel_type": ctype,
             "agent_id": str(agent_id).strip() if agent_id is not None else "",
+            "name": seeded_name,
             "credentials": incoming_creds,
         }
         cleaned = _clean_members(members, result_record["agent_id"])
@@ -429,6 +483,7 @@ def upsert_instance(
         instance_id=target_id,
         channel_type=ctype,
         agent_id=str(result_record.get("agent_id") or ""),
+        name=str(result_record.get("name") or ""),
         credentials=dict(result_record.get("credentials") or {}),
         members=list(result_record.get("members") or []),
         legacy=False,

channel/feishu/feishu_channel.py

@@ -569,7 +569,7 @@ class FeiShuChanel(ChatChannel):
 
         from agent.tools.scheduler.task_store import TaskStore
 
-        return TaskStore(str(state_dir.scheduler_file()))
+        return TaskStore(str(state_dir.scheduler_file_global()))
 
     def _send_scheduler_card(self, feishu_msg, is_group: bool, receive_id_type: str) -> bool:
         """Reply to ``/tasks`` with tasks scoped to the current chat."""

common/channel_registry.py

@@ -0,0 +1,26 @@
+"""Process-wide handle to the running ChannelManager.
+
+The entry module is launched as ``python app.py``, so it lives in
+``sys.modules['__main__']``. Any later ``import app`` then builds a *second*,
+distinct module object whose module-level globals (including ``_channel_mgr``)
+are separate from the ones ``run()`` populated. Code that reached for the live
+manager via ``from app import get_channel_manager`` therefore saw ``None`` and
+silently fell back to a bare, never-started channel singleton — the scheduler
+delivery bug behind issue #3120.
+
+Holding the manager here instead keeps a single shared cell: ``common`` is
+always imported as ``common.channel_registry`` (never as ``__main__``), so every
+caller — the entry module and the scheduler alike — reads and writes the same
+value.
+"""
+
+_channel_manager = None
+
+
+def set_channel_manager(manager) -> None:
+    global _channel_manager
+    _channel_manager = manager
+
+
+def get_channel_manager():
+    return _channel_manager

common/state_dir.py

@@ -15,8 +15,10 @@ State falls into three kinds, and which kind a path is decided here:
   drift; what actually differs between Agents is which ones are switched on,
   not which ones exist. An Agent that genuinely needs its own copy creates the
   directory and wins by presence.
-- **Per Agent** (``state_root``): persona, sessions, scheduled tasks, scratch.
-  What makes this Agent a different one from that Agent.
+- **Per Agent** (``state_root``): persona, sessions, scratch.
+  What makes this Agent a different one from that Agent. Scheduled tasks are
+  shared (each task carries its own ``agent_id``) so re-binding a channel
+  instance never has to move a file.
 - **Per end user** (``user_root``): profile, preferences, memory, task records.
   "Wang likes email over phone calls" is a fact about Wang, not about one
   Agent, so it sits beside the Agents rather than under one of them.
@@ -170,6 +172,38 @@ def env_file(identity=None, base=None) -> Path:
     return _shared_or_own(identity, base, ".env")
 
 
+def scheduler_recipients_file(base=None) -> Path:
+    """The one directory of people/groups observed on inbound IM channels.
+
+    Shared, not per Agent: a contact is a fact about the person, not about which
+    Agent happened to talk to them, so one copy lets any Agent's console build a
+    scheduled task for anyone the instance has ever met. It lives beside the
+    global ``tasks.json`` under ``scheduler`` — the recipients a scheduled push
+    can target — rather than in a separate directory.
+    """
+    if base is not None:
+        return Path(base) / "scheduler" / "recipients.json"
+    return shared_root() / "scheduler" / "recipients.json"
+
+
+def scheduler_file_global(base=None) -> Path:
+    """The one task store every Agent's scheduled tasks live in.
+
+    Shared, not per Agent, and each task carries its own ``agent_id`` (the Agent
+    it runs as) plus ``instance_id`` (the channel login it delivers through).
+    Keeping ownership *on the task* rather than *in the file path* is what lets a
+    channel instance be re-bound to a different Agent without moving any task
+    data: delivery resolves by ``instance_id`` and execution sets identity from
+    the task's ``agent_id``, neither of which depends on where the file sits.
+
+    Superseded the per-Agent ``scheduler_file`` (``<agent>/scheduler/tasks.json``);
+    a one-time boot migration folds those into this single store.
+    """
+    if base is not None:
+        return Path(base) / "scheduler" / "tasks.json"
+    return shared_root() / "scheduler" / "tasks.json"
+
+
 # --- Per Agent: what makes this Agent a different one -----------------------
 
 
@@ -177,11 +211,6 @@ def scheduler_file(identity=None, base=None) -> Path:
     return _agent_base(identity, base) / "scheduler" / "tasks.json"
 
 
-def scheduler_recipients_file(identity=None, base=None) -> Path:
-    """Trusted recipients observed on this Agent's inbound channels."""
-    return _agent_base(identity, base) / "scheduler" / "recipients.json"
-
-
 def tmp_dir(identity=None, ensure: bool = True, base=None) -> Path:
     """Transient downloads and synthesized media.
 

plugins/cow_cli/cow_cli.py

@@ -424,9 +424,9 @@ class CowCliPlugin(Plugin):
         task_store = get_task_store()
         if task_store is None:
             from agent.tools.scheduler.task_store import TaskStore
-            from common.state_dir import scheduler_file
+            from common.state_dir import scheduler_file_global
 
-            task_store = TaskStore(str(scheduler_file()))
+            task_store = TaskStore(str(scheduler_file_global()))
 
         channel_type = ""
         receiver = ""

tests/test_channel_instances.py

@@ -172,6 +172,60 @@ def test_upsert_honors_provided_id(tmp_path):
     assert inst.instance_id == "from-remote-123"
 
 
+# ---------------------------------------------------------------------------
+# instance name: a user-editable friendly label, seeded with a default and
+# fully optional so legacy records without one keep working.
+# ---------------------------------------------------------------------------
+
+def test_new_instance_gets_default_name(tmp_path):
+    settings = {"agent_workspace": str(tmp_path)}
+    first = ci.upsert_instance(settings, channel_type="weixin", credentials={})
+    second = ci.upsert_instance(settings, channel_type="weixin", credentials={})
+    # First of a type is unnumbered; the second is suffixed so they read apart.
+    assert first.name == "微信"
+    assert second.name == "微信 2"
+
+
+def test_upsert_can_set_and_clear_name(tmp_path):
+    settings = {"agent_workspace": str(tmp_path)}
+    inst = ci.upsert_instance(settings, channel_type="feishu",
+                              credentials={"feishu_app_id": "A"})
+    renamed = ci.upsert_instance(settings, channel_type="feishu",
+                                 instance_id=inst.instance_id, name="运营飞书")
+    assert renamed.name == "运营飞书"
+    # Renaming must not disturb credentials or duplicate the record.
+    assert renamed.credentials["feishu_app_id"] == "A"
+    assert len(ci.read_raw_instances(settings)) == 1
+    # Empty name clears the label back to unset.
+    cleared = ci.upsert_instance(settings, channel_type="feishu",
+                                 instance_id=inst.instance_id, name="")
+    assert cleared.name == ""
+
+
+def test_upsert_partial_update_keeps_name(tmp_path):
+    settings = {"agent_workspace": str(tmp_path)}
+    inst = ci.upsert_instance(settings, channel_type="feishu", name="客服飞书",
+                              credentials={"feishu_app_id": "A"})
+    # A credentials-only update (name=None) must preserve the existing label.
+    updated = ci.upsert_instance(settings, channel_type="feishu",
+                                 instance_id=inst.instance_id,
+                                 credentials={"feishu_app_secret": "S"})
+    assert updated.name == "客服飞书"
+
+
+def test_legacy_record_without_name_resolves(tmp_path):
+    # A record written before the name field existed must still load, with an
+    # empty name (the UI falls back to the id) and no error.
+    settings = _write_instances(
+        tmp_path,
+        [{"instance_id": "feishu-old", "channel_type": "feishu",
+          "credentials": {"feishu_app_id": "A"}}],
+    )
+    inst = ci.resolve_channel_instances(settings)[0]
+    assert inst.name == ""
+    assert inst.instance_id == "feishu-old"
+
+
 def test_remove_instance(tmp_path):
     settings = {"agent_workspace": str(tmp_path)}
     inst = ci.upsert_instance(settings, channel_type="feishu",

tests/test_multi_agent_state_isolation.py

@@ -119,31 +119,79 @@ def test_memory_config_keeps_each_agent_index_under_its_workspace(
     assert research_db == Path(research.workspace) / "memory/long-term/index.db"
 
 
-def test_scheduler_stores_allow_same_task_id_per_agent(isolated_registry):
+def test_scheduler_is_one_global_store_tagged_by_agent_id(isolated_registry):
+    """Tasks for every Agent live in one file; ownership is a field, not a path.
+    Re-binding a channel instance therefore never has to move a task."""
     class Bridge:
         agent_registry = isolated_registry
 
     bridge = Bridge()
-    for profile in isolated_registry.list(include_disabled=False):
-        assert init_scheduler(bridge, profile.workspace, profile.id)
+    assert init_scheduler(bridge)
 
-    primary_store = get_task_store(agent_id="primary")
-    research_store = get_task_store(agent_id="research")
-    primary_store.add_task(_task("daily", "Primary daily"))
-    research_store.add_task(_task("daily", "Research daily"))
+    store = get_task_store()
+    primary = _task("daily-p", "Primary daily")
+    primary["agent_id"] = "primary"
+    research = _task("daily-r", "Research daily")
+    research["agent_id"] = "research"
+    store.add_task(primary)
+    store.add_task(research)
 
-    assert primary_store is not research_store
-    assert primary_store.get_task("daily")["name"] == "Primary daily"
-    assert research_store.get_task("daily")["name"] == "Research daily"
-    assert Path(primary_store.store_path) == Path(
-        isolated_registry.get("primary").workspace
-    ) / "scheduler/tasks.json"
-    assert Path(research_store.store_path) == Path(
-        isolated_registry.get("research").workspace
-    ) / "scheduler/tasks.json"
-    assert get_scheduler_service(agent_id="primary") is not get_scheduler_service(
+    assert get_task_store(agent_id="primary") is store
+    assert get_task_store(agent_id="research") is store
+    assert get_scheduler_service(agent_id="primary") is get_scheduler_service(
         agent_id="research"
     )
+    assert [t["id"] for t in store.list_tasks(agent_id="primary")] == ["daily-p"]
+    assert [t["id"] for t in store.list_tasks(agent_id="research")] == ["daily-r"]
+    assert Path(store.store_path) == Path(
+        isolated_registry.get("primary").workspace
+    ) / "scheduler/tasks.json"
+
+
+def test_legacy_per_agent_task_files_fold_into_the_global_store(isolated_registry):
+    """A one-time boot migration imports each Agent's old tasks.json, stamps
+    ``agent_id``, and leaves the source renamed aside — not deleted, so a
+    rollback is possible. The default Agent's file *is* the global store, so
+    those tasks stay put and only get an owner stamp."""
+    import json
+
+    class Bridge:
+        agent_registry = isolated_registry
+
+    primary = isolated_registry.get("primary")
+    research = isolated_registry.get("research")
+    global_path = Path(primary.workspace) / "scheduler" / "tasks.json"
+    legacy_path = Path(research.workspace) / "scheduler" / "tasks.json"
+    global_path.parent.mkdir(parents=True, exist_ok=True)
+    legacy_path.parent.mkdir(parents=True, exist_ok=True)
+    global_path.write_text(
+        json.dumps({"version": 1, "tasks": {"old-p": _task("old-p", "Default leftover")}}),
+        encoding="utf-8",
+    )
+    legacy_path.write_text(
+        json.dumps({"version": 1, "tasks": {"old-r": _task("old-r", "Research leftover")}}),
+        encoding="utf-8",
+    )
+
+    assert init_scheduler(Bridge())
+    store = get_task_store()
+    assert store.get_task("old-p")["agent_id"] == "primary"
+    assert store.get_task("old-r")["agent_id"] == "research"
+    assert store.get_task("old-r")["name"] == "Research leftover"
+    assert not legacy_path.exists()
+    assert legacy_path.with_name("tasks.json.migrated").exists()
+
+
+def test_list_tasks_treats_missing_agent_id_as_the_default(isolated_registry):
+    class Bridge:
+        agent_registry = isolated_registry
+
+    assert init_scheduler(Bridge())
+    store = get_task_store()
+    store.add_task(_task("orphan", "No owner field"))
+
+    assert [t["id"] for t in store.list_tasks(agent_id="primary")] == ["orphan"]
+    assert store.list_tasks(agent_id="research") == []
 
 
 def test_each_agent_boots_only_its_own_mcp_servers(mcp_workspaces, monkeypatch):

tests/test_scheduler_channel_resolution.py

@@ -0,0 +1,295 @@
+"""Regression tests for scheduler channel resolution / readiness (issue #3120).
+
+A weixin bot that runs as a channel_instance registers under its instance_id
+(e.g. ``weixin-c696...``). A scheduled task whose action never stored an
+instance_id must still resolve to that live instance instead of a bare,
+never-started ``create_channel`` singleton, otherwise its token map is empty and
+readiness defers the task forever.
+"""
+
+import sys
+import types
+
+import agent.tools.scheduler.integration as integration
+
+
+class _FakeChannel:
+    def __init__(self, channel_type, instance_id="", tokens=None):
+        self.channel_type = channel_type
+        self.instance_id = instance_id
+        self._context_tokens = dict(tokens or {})
+
+
+class _FakeManager:
+    def __init__(self, channels):
+        # channels: dict registry_name -> _FakeChannel
+        self._channels = dict(channels)
+
+    def get_channel(self, name):
+        return self._channels.get(name)
+
+    def find_channels_by_type(self, channel_type):
+        return [
+            (name, ch)
+            for name, ch in self._channels.items()
+            if getattr(ch, "channel_type", "") == channel_type
+        ]
+
+
+def _install_manager(monkeypatch, manager):
+    """Make ``common.channel_registry.get_channel_manager`` return our fake."""
+    import common.channel_registry as registry
+
+    monkeypatch.setattr(registry, "get_channel_manager", lambda: manager)
+
+
+class _FakeRecipientStore:
+    def __init__(self, entries):
+        self._entries = list(entries)
+
+    def list(self):
+        return list(self._entries)
+
+
+def _install_directory(monkeypatch, entries):
+    monkeypatch.setattr(
+        integration,
+        "_get_shared_recipient_store",
+        lambda: _FakeRecipientStore(entries),
+    )
+
+
+def test_resolves_live_instance_when_task_has_no_instance_id(monkeypatch):
+    live = _FakeChannel("weixin", "weixin-c6961093a5", tokens={"user@im": "tok"})
+    _install_manager(monkeypatch, _FakeManager({"weixin-c6961093a5": live}))
+    _install_directory(monkeypatch, [])
+
+    resolved = integration._resolve_delivery_channel("weixin", "", "user@im")
+
+    assert resolved is live
+
+
+def test_ready_true_for_instance_run_weixin_without_instance_id(monkeypatch):
+    live = _FakeChannel("weixin", "weixin-c6961093a5", tokens={"user@im": "tok"})
+    _install_manager(monkeypatch, _FakeManager({"weixin-c6961093a5": live}))
+    _install_directory(monkeypatch, [])
+
+    assert integration._is_channel_ready("weixin", "user@im", instance_id="") is True
+
+
+def test_multiple_instances_prefers_the_one_holding_the_receiver(monkeypatch):
+    other = _FakeChannel("weixin", "weixin-aaaa", tokens={"someone-else": "t"})
+    mine = _FakeChannel("weixin", "weixin-bbbb", tokens={"user@im": "tok"})
+    _install_manager(
+        monkeypatch,
+        _FakeManager({"weixin-aaaa": other, "weixin-bbbb": mine}),
+    )
+    _install_directory(monkeypatch, [])  # no directory hit -> fall to token match
+
+    resolved = integration._resolve_delivery_channel("weixin", "", "user@im")
+
+    assert resolved is mine
+
+
+def test_feishu_recovers_correct_instance_from_recipient_directory(monkeypatch):
+    """Two Feishu apps; a task with no instance_id must resolve to the app that
+    actually owns the open_id, not the first one (which raises open_id cross
+    app). The recipient directory is the authoritative source of that mapping."""
+    app_a = _FakeChannel("feishu", "feishu-72734a6dbe")
+    app_b = _FakeChannel("feishu", "feishu-ec75a1f1f8")
+    _install_manager(
+        monkeypatch,
+        _FakeManager({"feishu-72734a6dbe": app_a, "feishu-ec75a1f1f8": app_b}),
+    )
+    _install_directory(
+        monkeypatch,
+        [
+            {
+                "channel_type": "feishu",
+                "instance_id": "feishu-ec75a1f1f8",
+                "receiver": "ou_a2a4af",
+            }
+        ],
+    )
+
+    resolved = integration._resolve_delivery_channel("feishu", "", "ou_a2a4af")
+
+    assert resolved is app_b
+
+
+def test_feishu_refuses_to_guess_between_instances_without_a_directory_hit(monkeypatch):
+    """No instance_id and the receiver is unknown to the directory: we must not
+    blindly deliver through the first app (that is the cross-app bug). Falling
+    through to create_channel is fine — the point is we don't return app_a."""
+    app_a = _FakeChannel("feishu", "feishu-72734a6dbe")
+    app_b = _FakeChannel("feishu", "feishu-ec75a1f1f8")
+    _install_manager(
+        monkeypatch,
+        _FakeManager({"feishu-72734a6dbe": app_a, "feishu-ec75a1f1f8": app_b}),
+    )
+    _install_directory(monkeypatch, [])
+
+    bare = object()
+    fake_factory = types.ModuleType("channel.channel_factory")
+    fake_factory.create_channel = lambda *a, **k: bare
+    monkeypatch.setitem(sys.modules, "channel.channel_factory", fake_factory)
+
+    resolved = integration._resolve_delivery_channel("feishu", "", "ou_unknown")
+
+    assert resolved is not app_a
+    assert resolved is not app_b
+    assert resolved is bare
+
+
+def test_single_feishu_instance_is_used_directly(monkeypatch):
+    """Solo Feishu install: exactly one instance, so it is safe to auto-pick
+    even without a directory entry (legacy behaviour preserved)."""
+    only = _FakeChannel("feishu", "feishu-72734a6dbe")
+    _install_manager(monkeypatch, _FakeManager({"feishu-72734a6dbe": only}))
+    _install_directory(monkeypatch, [])
+
+    resolved = integration._resolve_delivery_channel("feishu", "", "ou_a2a4af")
+
+    assert resolved is only
+
+
+def test_manager_is_shared_across_app_main_duplication(monkeypatch):
+    """app.py runs as __main__; a later ``import app`` builds a second module.
+    The manager must be held in the shared registry so the scheduler (which
+    imports it, not __main__) sees the same object the entry module set — the
+    core of issue #3120."""
+    import common.channel_registry as registry
+
+    sentinel = object()
+    monkeypatch.setattr(registry, "_channel_manager", None)
+    try:
+        # Whatever the entry module set...
+        registry.set_channel_manager(sentinel)
+        # ...the scheduler-side import must observe.
+        assert registry.get_channel_manager() is sentinel
+    finally:
+        registry.set_channel_manager(None)
+
+
+def test_exact_instance_id_still_wins(monkeypatch):
+    a = _FakeChannel("weixin", "weixin-aaaa", tokens={"user@im": "t"})
+    b = _FakeChannel("weixin", "weixin-bbbb", tokens={"user@im": "t"})
+    _install_manager(monkeypatch, _FakeManager({"weixin-aaaa": a, "weixin-bbbb": b}))
+
+    resolved = integration._resolve_delivery_channel("weixin", "weixin-bbbb", "user@im")
+
+    assert resolved is b
+
+
+def test_bare_singleton_fallback_does_not_hard_block_readiness(monkeypatch):
+    """No live instance of the type: the token check is inconclusive, so the
+    scheduler must allow the send attempt instead of deferring forever."""
+    _install_manager(monkeypatch, _FakeManager({}))
+
+    bare = _FakeChannel("weixin", "", tokens={})
+    monkeypatch.setattr(
+        integration, "_resolve_delivery_channel", lambda *a, **k: bare
+    )
+
+    assert integration._is_channel_ready("weixin", "user@im", instance_id="") is True
+
+
+def test_live_instance_without_token_is_still_not_ready(monkeypatch):
+    """A real, running instance that has never seen the receiver is genuinely
+    not ready — we should defer, not blast into a login that can't route it."""
+    live = _FakeChannel("weixin", "weixin-c6961093a5", tokens={})
+    _install_manager(monkeypatch, _FakeManager({"weixin-c6961093a5": live}))
+    _install_directory(monkeypatch, [])
+
+    assert integration._is_channel_ready("weixin", "user@im", instance_id="") is False
+
+
+# ── Backward-compatibility: legacy tasks with no instance_id / no agent_id ──
+#
+# Pre-existing tasks (created before multi-instance) store neither instance_id
+# nor a per-instance login. They MUST keep delivering exactly as before. These
+# tests pin that contract so a future change can't quietly break old data.
+
+
+def test_legacy_single_instance_registers_under_channel_type(monkeypatch):
+    """A non-multi install runs its channel under the bare channel_type key
+    (``weixin``). A legacy task with no instance_id resolves to it via
+    ``lookup = instance_id or channel_type`` — the original single-bot path."""
+    legacy = _FakeChannel("weixin", "weixin", tokens={"user@im": "tok"})
+    _install_manager(monkeypatch, _FakeManager({"weixin": legacy}))
+    _install_directory(monkeypatch, [])
+
+    resolved = integration._resolve_delivery_channel("weixin", "", "user@im")
+
+    assert resolved is legacy
+    assert integration._is_channel_ready("weixin", "user@im", instance_id="") is True
+
+
+def test_no_manager_falls_back_to_create_channel_like_before(monkeypatch):
+    """Before ChannelManager exists (or any failure reaching it), resolution
+    must fall back to ``create_channel(channel_type)`` — byte-for-byte the old
+    behaviour, so a solo install and old tasks are untouched."""
+    import common.channel_registry as registry
+
+    monkeypatch.setattr(registry, "get_channel_manager", lambda: None)
+
+    bare = object()
+    fake_factory = types.ModuleType("channel.channel_factory")
+    called = {}
+
+    def _create(channel_type, *a, **k):
+        called["channel_type"] = channel_type
+        return bare
+
+    fake_factory.create_channel = _create
+    monkeypatch.setitem(sys.modules, "channel.channel_factory", fake_factory)
+
+    resolved = integration._resolve_delivery_channel("feishu", "", "ou_x")
+
+    assert resolved is bare
+    assert called["channel_type"] == "feishu"
+
+
+def test_legacy_feishu_task_without_instance_id_single_instance(monkeypatch):
+    """The common legacy case: one Feishu app, an old task with no instance_id
+    and no directory entry. Auto-picking the sole running instance is safe and
+    keeps the task delivering."""
+    only = _FakeChannel("feishu", "feishu")
+    _install_manager(monkeypatch, _FakeManager({"feishu": only}))
+    _install_directory(monkeypatch, [])
+
+    resolved = integration._resolve_delivery_channel("feishu", "", "ou_f1cbd4")
+
+    assert resolved is only
+
+
+def test_comma_joined_channel_type_is_normalized(monkeypatch):
+    """config.json permits ``channel_type='web,feishu'``; a task copied from it
+    stores that whole string. Readiness/resolution must not choke on it — the
+    primary type is taken so the task still resolves instead of raising."""
+    assert integration._primary_channel_type("feishu,dingtalk") == "feishu"
+    assert integration._primary_channel_type("") == "unknown"
+    assert integration._primary_channel_type(None) == "unknown"
+
+
+def test_unknown_channel_type_is_always_ready(monkeypatch):
+    """A task with a missing/unknown channel_type must not be blocked by the
+    readiness probe (unknown -> True), preserving pre-check behaviour."""
+    assert integration._is_channel_ready("", "whoever") is True
+    assert integration._is_channel_ready("unknown", "whoever") is True
+
+
+def test_directory_lookup_failure_does_not_break_resolution(monkeypatch):
+    """If the recipient directory is unreadable, resolution must degrade
+    gracefully (fall through), never raise into the scheduler tick."""
+    only = _FakeChannel("feishu", "feishu")
+    _install_manager(monkeypatch, _FakeManager({"feishu": only}))
+
+    def _boom():
+        raise OSError("directory unavailable")
+
+    monkeypatch.setattr(integration, "_get_shared_recipient_store", _boom)
+
+    # Single instance -> still resolves despite the directory blowing up.
+    resolved = integration._resolve_delivery_channel("feishu", "", "ou_f1cbd4")
+    assert resolved is only

tests/test_scheduler_task_ownership.py

@@ -0,0 +1,127 @@
+"""Ownership of a scheduled task follows its delivery channel instance.
+
+A channel instance binds to exactly one Agent. An IM task delivers *through* an
+instance, so the Agent it runs as (and the owner the console shows) is derived
+live from that instance's current binding — never a value frozen on the task.
+This is what lets an instance be re-bound to another Agent with zero task
+migration, and keeps the edit form / task card honest the moment the delivery
+instance is switched.
+"""
+
+import types
+
+import agent.tools.scheduler.integration as integration
+from agent.tools.scheduler.integration import effective_task_agent_id
+
+
+class _FakeInstance:
+    def __init__(self, agent_id):
+        self.agent_id = agent_id
+
+
+class _FakeRegistry:
+    """Minimal registry: knows a fixed set of enabled agent ids + a default."""
+
+    def __init__(self, agents, default_id):
+        self._agents = set(agents)
+        self.default_agent_id = default_id
+
+    def get(self, agent_id, require_enabled=False):
+        if agent_id in self._agents:
+            return types.SimpleNamespace(id=agent_id)
+        raise KeyError(agent_id)
+
+
+def _bind_instances(monkeypatch, mapping):
+    """Point ``get_instance`` at a fixed instance_id -> agent_id map."""
+    import channel.channel_instances as ci
+
+    def _get_instance(_settings, instance_id):
+        agent_id = mapping.get(instance_id)
+        return _FakeInstance(agent_id) if agent_id is not None else None
+
+    monkeypatch.setattr(ci, "get_instance", _get_instance)
+    monkeypatch.setattr("config.conf", lambda: {}, raising=False)
+
+
+def _im_task(instance_id, channel_type="feishu", stored_agent_id=""):
+    return {
+        "id": "t1",
+        "agent_id": stored_agent_id,
+        "action": {
+            "type": "agent_task",
+            "task_description": "x",
+            "channel_type": channel_type,
+            "instance_id": instance_id,
+            "receiver": "u1",
+        },
+    }
+
+
+def test_im_task_owner_follows_the_instance_not_the_stored_value(monkeypatch):
+    # The instance is bound to "beta"; the task still carries a stale "alpha".
+    _bind_instances(monkeypatch, {"feishu-1": "beta"})
+    registry = _FakeRegistry({"alpha", "beta"}, default_id="alpha")
+    task = _im_task("feishu-1", stored_agent_id="alpha")
+
+    assert effective_task_agent_id(task, registry) == "beta"
+
+
+def test_rebinding_the_instance_moves_the_task_with_no_rewrite(monkeypatch):
+    registry = _FakeRegistry({"alpha", "beta"}, default_id="alpha")
+    task = _im_task("feishu-1", stored_agent_id="alpha")
+
+    _bind_instances(monkeypatch, {"feishu-1": "alpha"})
+    assert effective_task_agent_id(task, registry) == "alpha"
+
+    # Re-bind the same instance; the task dict is untouched but ownership flips.
+    _bind_instances(monkeypatch, {"feishu-1": "beta"})
+    assert effective_task_agent_id(task, registry) == "beta"
+    assert task["agent_id"] == "alpha"  # storage never rewritten
+
+
+def test_unbound_instance_falls_back_to_the_stored_owner(monkeypatch):
+    # Instance exists but has no explicit binding (agent_id empty) -> stored wins.
+    _bind_instances(monkeypatch, {"feishu-1": ""})
+    registry = _FakeRegistry({"alpha", "beta"}, default_id="alpha")
+    task = _im_task("feishu-1", stored_agent_id="beta")
+
+    assert effective_task_agent_id(task, registry) == "beta"
+
+
+def test_web_task_keeps_its_stored_owner(monkeypatch):
+    _bind_instances(monkeypatch, {})
+    registry = _FakeRegistry({"alpha", "beta"}, default_id="alpha")
+    task = {
+        "id": "w1",
+        "agent_id": "beta",
+        "action": {"type": "send_message", "channel_type": "web", "receiver": "s1"},
+    }
+
+    assert effective_task_agent_id(task, registry) == "beta"
+
+
+def test_execution_degrades_to_default_when_instance_binds_disabled_agent(monkeypatch):
+    # Instance binds "ghost", who is not an enabled Agent. Execution must not run
+    # as a phantom identity: validate=True degrades to the default.
+    _bind_instances(monkeypatch, {"feishu-1": "ghost"})
+    registry = _FakeRegistry({"alpha"}, default_id="alpha")
+    task = _im_task("feishu-1", stored_agent_id="alpha")
+
+    class _Bridge:
+        agent_registry = registry
+
+    assert integration._resolve_task_agent_id(_Bridge(), task) == "alpha"
+
+
+def test_display_keeps_stored_owner_without_a_live_registry(monkeypatch):
+    # Display path (validate=False) tolerates a missing registry: the owner is a
+    # label, so a task keeps showing its stored agent_id in bare contexts.
+    _bind_instances(monkeypatch, {})
+    task = {
+        "id": "n1",
+        "agent_id": "research",
+        "action": {"type": "agent_task", "task_description": "x"},
+    }
+
+    assert effective_task_agent_id(task, registry=None) == "research"