Merge pull request #6065 from MervinPraison/claude/issue-6037-20261026

fix: deliver background subagent results to the originating chat and list them in /tasks

src/praisonai-agents/praisonaiagents/tools/subagent_tool.py

@@ -66,6 +66,35 @@ def _emit_subagent_stop(
         logger.debug("SUBAGENT_STOP hook failed", exc_info=True)
 
 
+def _messenger_delivery(messenger: Any) -> Callable[[Any], None]:
+    """Build an ``on_complete`` callback that sends a finished job's result
+    back to the turn's own chat (``"origin"``) through its outbound messenger.
+
+    Raises when the send is not accepted so the job manager leaves the job
+    un-``delivered`` instead of recording a delivery that never happened.
+    """
+
+    def _deliver(info: Any) -> None:
+        from .messaging_tools import _run_async
+
+        result = info.result
+        if info.error:
+            text = f"Background task failed: {info.error}"
+        elif isinstance(result, dict):
+            text = result.get("output") if result.get("success") else (
+                f"Background task failed: {result.get('error')}"
+            )
+        else:
+            text = result
+        if not text:
+            text = "Background task finished with no output."
+        outcome = _run_async(messenger.send("origin", str(text)))
+        if not getattr(outcome, "ok", False):
+            raise RuntimeError(getattr(outcome, "summary", "") or "delivery failed")
+
+    return _deliver
+
+
 def create_subagent_tool(
     agent_factory: Optional[Callable[..., Any]] = None,
     allowed_agents: Optional[List[str]] = None,
@@ -114,9 +143,11 @@ def create_subagent_tool(
             ``deliver`` target finishes. This is the observable completion
             signal a gateway (e.g. ``BotOS``) subscribes to in order to
             route the result back to the originating chat without an active
-            turn. When ``None`` (the default) background jobs remain strictly
-            pull-only via ``subagent_result`` — byte-for-byte the prior
-            behaviour.
+            turn. When ``None`` (the default), ``deliver="origin"`` targets
+            the turn's own chat, the spawning turn has an outbound messenger
+            registered (a running gateway) and the send policy allows it, the
+            result is sent back through that messenger; otherwise background
+            jobs remain pull-only via ``subagent_result``.
         
     Returns:
         Tool definition dictionary
@@ -356,16 +387,45 @@ def create_subagent_tool(
             origin: Dict[str, Any] = {}
             job_on_complete = None
             if deliver:
+                # Fill the origin from the current turn (still in scope here)
+                # rather than from the model; explicit arguments still win.
+                from praisonaiagents.session.context import (
+                    get_outbound_messenger,
+                    get_session_context,
+                )
+
+                ctx = get_session_context()
+                use_ctx = not chat_id or chat_id == ctx.chat_id
                 origin = {
                     "deliver": deliver,
-                    "platform": platform,
-                    "chat_id": chat_id,
-                    "thread_id": thread_id,
+                    "platform": platform or (ctx.platform if use_ctx else ""),
+                    "chat_id": chat_id or ctx.chat_id,
+                    "thread_id": thread_id or (ctx.thread_id if use_ctx else ""),
                     "session_id": session_id,
+                    "user_id": ctx.user_id,
                     "task": task,
                 }
                 if on_job_complete is not None:
                     job_on_complete = on_job_complete
+                else:
+                    # Auto-send only back to the turn's own chat and only if
+                    # the send policy allows it (as ``send_message`` does);
+                    # any other target needs an explicit ``on_job_complete``.
+                    from .messaging_tools import _check_send_policy
+
+                    messenger = get_outbound_messenger()
+                    same_chat = (
+                        (not chat_id or chat_id == ctx.chat_id)
+                        and (not platform or platform == ctx.platform)
+                        and (not thread_id or thread_id == ctx.thread_id)
+                    )
+                    if (
+                        messenger is not None
+                        and deliver == "origin"
+                        and same_chat
+                        and _check_send_policy(deliver) is None
+                    ):
+                        job_on_complete = _messenger_delivery(messenger)
 
             job_id = job_manager.start_job(
                 _job,

src/praisonai-agents/tests/unit/tools/test_subagent_origin_context.py

@@ -0,0 +1,178 @@
+"""spawn_subagent(background=True, deliver=...) inside a gateway turn (#6037).
+
+The model cannot supply the origin chat, so it must come from the turn's
+session context, and the result must reach the turn's outbound messenger.
+"""
+
+import time
+
+from praisonaiagents.background.job_manager import get_job_manager
+from praisonaiagents.gateway.protocols import DeliveryResult
+from praisonaiagents.session.context import (
+    clear_outbound_messenger,
+    clear_session_context,
+    register_outbound_messenger,
+    set_session_context,
+)
+from praisonaiagents.tools.subagent_tool import create_subagent_tool
+
+
+class _StubAgent:
+    def chat(self, prompt):
+        return "long job done"
+
+
+class _FakeMessenger:
+    def __init__(self, ok=True):
+        self.ok = ok
+        self.sent = []
+
+    async def send(self, target, text, *, media=None):
+        self.sent.append((target, text))
+        return DeliveryResult(ok=self.ok, target=target)
+
+    def list_targets(self):
+        return []
+
+
+def _spawn_in_turn(messenger, deliver="origin", **kwargs):
+    tool = create_subagent_tool(agent_factory=lambda **kw: _StubAgent())
+    ctx = set_session_context(
+        platform="telegram", chat_id="100", thread_id="7", user_id="alice"
+    )
+    msg = register_outbound_messenger(messenger)
+    try:
+        return tool["function"](
+            task="summarise", background=True, deliver=deliver, **kwargs
+        )
+    finally:
+        clear_outbound_messenger(msg)
+        clear_session_context(ctx)
+
+
+def _wait_info(job_id):
+    jm = get_job_manager()
+    jm.get_result(job_id, timeout=5)
+    for _ in range(50):
+        info = jm.get_job_info(job_id)
+        if info.delivered:
+            break
+        time.sleep(0.02)
+    return info
+
+
+def test_origin_filled_from_session_context_and_delivered_once():
+    messenger = _FakeMessenger()
+    handle = _spawn_in_turn(messenger)
+    info = _wait_info(handle["job_id"])
+
+    assert info.origin["platform"] == "telegram"
+    assert info.origin["chat_id"] == "100"
+    assert info.origin["thread_id"] == "7"
+    assert info.origin["user_id"] == "alice"
+    assert info.delivered is True
+    assert len(messenger.sent) == 1
+    target, text = messenger.sent[0]
+    assert target == "origin"
+    assert "long job done" in text
+
+
+def test_explicit_origin_arguments_win():
+    messenger = _FakeMessenger()
+    handle = _spawn_in_turn(messenger, platform="slack", chat_id="C9")
+    info = _wait_info(handle["job_id"])
+    assert info.origin["platform"] == "slack"
+    assert info.origin["chat_id"] == "C9"
+    assert info.origin["thread_id"] == ""
+    assert info.origin["user_id"] == "alice"
+    # A different chat is not the turn's chat: never send it to chat "100".
+    assert messenger.sent == []
+    assert info.delivered is False
+
+
+def test_explicit_other_thread_is_not_auto_sent():
+    messenger = _FakeMessenger()
+    handle = _spawn_in_turn(messenger, thread_id="99")
+    info = _wait_info(handle["job_id"])
+    assert info.origin["thread_id"] == "99"
+    assert messenger.sent == []
+    assert info.delivered is False
+
+
+def test_non_origin_target_is_not_auto_sent():
+    messenger = _FakeMessenger()
+    handle = _spawn_in_turn(messenger, deliver="telegram:999")
+    info = _wait_info(handle["job_id"])
+    assert messenger.sent == []
+    assert info.delivered is False
+
+
+def test_send_policy_denial_blocks_auto_send():
+    from praisonaiagents.gateway.protocols import SendDecision
+    from praisonaiagents.session.context import (
+        clear_send_policy,
+        register_send_policy,
+    )
+
+    class _Deny:
+        def evaluate(self, target, **kw):
+            return SendDecision(allow=False, reason="no")
+
+    messenger = _FakeMessenger()
+    token = register_send_policy(_Deny())
+    try:
+        handle = _spawn_in_turn(messenger)
+    finally:
+        clear_send_policy(token)
+    info = _wait_info(handle["job_id"])
+    assert messenger.sent == []
+    assert info.delivered is False
+
+
+def test_rejected_send_is_not_marked_delivered():
+    messenger = _FakeMessenger(ok=False)
+    handle = _spawn_in_turn(messenger)
+    jm = get_job_manager()
+    jm.get_result(handle["job_id"], timeout=5)
+    time.sleep(0.1)
+    assert len(messenger.sent) == 1
+    assert jm.get_job_info(handle["job_id"]).delivered is False
+
+
+def test_without_gateway_stays_pull_only():
+    tool = create_subagent_tool(agent_factory=lambda **kw: _StubAgent())
+    handle = tool["function"](task="t", background=True, deliver="origin")
+    jm = get_job_manager()
+    jm.get_result(handle["job_id"], timeout=5)
+    time.sleep(0.05)
+    info = jm.get_job_info(handle["job_id"])
+    assert info.delivered is False
+    assert info.origin["chat_id"] == ""
+
+
+def test_restating_own_chat_keeps_turn_thread_and_platform():
+    messenger = _FakeMessenger()
+    handle = _spawn_in_turn(messenger, chat_id="100")
+    info = _wait_info(handle["job_id"])
+    assert info.origin["platform"] == "telegram"
+    assert info.origin["thread_id"] == "7"
+    assert len(messenger.sent) == 1
+
+
+def test_empty_output_still_notifies_before_marking_delivered():
+    class _Silent:
+        def chat(self, prompt):
+            return ""
+
+    tool = create_subagent_tool(agent_factory=lambda **kw: _Silent())
+    messenger = _FakeMessenger()
+    ctx = set_session_context(platform="telegram", chat_id="100", user_id="alice")
+    msg = register_outbound_messenger(messenger)
+    try:
+        handle = tool["function"](task="t", background=True, deliver="origin")
+    finally:
+        clear_outbound_messenger(msg)
+        clear_session_context(ctx)
+    info = _wait_info(handle["job_id"])
+    assert info.delivered is True
+    assert len(messenger.sent) == 1

src/praisonai-bot/praisonai_bot/bots/_commands.py

@@ -1475,6 +1475,7 @@ def handle_tasks_command(
     user_id: str,
     args: Optional[str] = None,
     runner: Optional[Any] = None,
+    job_manager: Optional[Any] = None,
 ) -> str:
     """Handle /tasks to inspect background tasks from inside a chat session.
 
@@ -1492,6 +1493,9 @@ def handle_tasks_command(
         user_id: The requesting user's identifier (owner scope).
         args: Optional argument string (an id, or ``cancel <id>``).
         runner: Optional ``BackgroundRunner`` (defaults to the shared one).
+        job_manager: Optional ``BackgroundJobManager`` holding background
+            subagents (defaults to the shared one). Jobs are owned through
+            ``origin["user_id"]``, captured when they were spawned.
 
     Returns:
         A compact, chat-friendly text response.
@@ -1504,6 +1508,28 @@ def handle_tasks_command(
         except Exception as e:  # noqa: BLE001
             return f"❌ Background tasks unavailable: {e}"
 
+    if job_manager is None:
+        try:
+            from praisonaiagents.background.job_manager import get_job_manager
+
+            job_manager = get_job_manager()
+        except Exception:  # noqa: BLE001
+            job_manager = None
+
+    def _mine(info: Any) -> bool:
+        # Same fail-closed rule as ``_owned``: jobs without an owner are hidden.
+        owner = (info.origin or {}).get("user_id")
+        return bool(owner) and str(owner) == str(user_id)
+
+    def _owned_job(job_id: str) -> Optional[Any]:
+        if job_manager is None:
+            return None
+        try:
+            info = job_manager.get_job_info(job_id)
+        except Exception:  # noqa: BLE001
+            return None
+        return info if _mine(info) else None
+
     def _owned(task: Dict[str, Any]) -> bool:
         # Fail closed: a task is only visible/actionable to the requesting user
         # when its owner id matches exactly. Ownerless tasks (e.g. submitted
@@ -1525,7 +1551,11 @@ def handle_tasks_command(
         except Exception:  # noqa: BLE001
             task = None
         if task is None or not _owned(task.to_dict()):
-            return f"❌ Task not found: {task_id}"
+            if _owned_job(task_id) is None:
+                return f"❌ Task not found: {task_id}"
+            if not job_manager.cancel_job(task_id):
+                return f"❌ Could not cancel task {task_id} (already finished?)."
+            return f"✅ Cancelled task {task_id}."
         try:
             # Cancel through the runner so the underlying future is actually
             # stopped (not just the record marked cancelled). cancel_task_sync
@@ -1549,7 +1579,21 @@ def handle_tasks_command(
         except Exception:  # noqa: BLE001
             task = None
         if task is None or not _owned(task.to_dict()):
-            return f"❌ Task not found: {arg}"
+            info = _owned_job(arg)
+            if info is None:
+                return f"❌ Task not found: {arg}"
+            lines = [
+                f"🧩 Task {info.job_id} — {info.origin.get('task') or 'subagent'}",
+                f"Status: {info.status.value}",
+            ]
+            if info.error:
+                lines.append(f"Error: {str(info.error)[:200]}")
+            elif info.result is not None:
+                result = info.result
+                if isinstance(result, dict):
+                    result = result.get("output") or result.get("error")
+                lines.append(f"Result: {str(result)[:200]}")
+            return "\n".join(lines)
         t = task.to_dict()
         lines = [
             f"🧩 Task {t.get('id')} — {t.get('name') or 'unnamed'}",
@@ -1566,8 +1610,19 @@ def handle_tasks_command(
         tasks = [t for t in runner.list_tasks() if _owned(t)]
     except Exception as e:  # noqa: BLE001
         return f"❌ Could not list tasks: {e}"
+    try:
+        jobs = sorted(
+            (
+                j for j in (job_manager.list_jobs().values() if job_manager else ())
+                if _mine(j)
+            ),
+            key=lambda j: j.created_at,
+            reverse=True,
+        )
+    except Exception:  # noqa: BLE001
+        jobs = []
 
-    if not tasks:
+    if not tasks and not jobs:
         return "💤 No background tasks."
 
     lines = ["🧩 Background tasks:"]
@@ -1577,6 +1632,9 @@ def handle_tasks_command(
             f"• {t.get('id')} {t.get('name') or 'unnamed'} — "
             f"{t.get('status')} ({progress})"
         )
+    for j in jobs[: max(0, 20 - len(tasks))]:
+        name = str(j.origin.get("task") or "subagent")[:60]
+        lines.append(f"• {j.job_id} {name} — {j.status.value}")
     lines.append("\nUse /tasks <id> for detail or /tasks cancel <id>.")
     return "\n".join(lines)
 

src/praisonai-bot/praisonai_bot/bots/_session.py

@@ -1325,8 +1325,14 @@ class BotSessionManager:
                 from praisonaiagents.session.context import (
                     register_outbound_messenger as _register_messenger,
                     clear_outbound_messenger as _clear_messenger,
+                    register_gateway_loop,
                 )
 
+                # Background workers (e.g. a ``deliver``-backed subagent)
+                # send after the turn ends; record the loop the router's
+                # adapters live on so they schedule onto it.
+                register_gateway_loop(asyncio.get_running_loop())
+
                 origin_source = None
                 if self._platform and chat_id:
                     origin_source = SessionSource(

src/praisonai-bot/tests/unit/bots/test_tasks_command_job_manager.py

@@ -0,0 +1,75 @@
+"""/tasks over BackgroundJobManager jobs (background subagents, #6037)."""
+
+import threading
+
+from praisonaiagents.background import BackgroundRunner
+from praisonaiagents.background.job_manager import BackgroundJobManager
+
+from praisonai_bot.bots._commands import handle_tasks_command
+
+
+def _start(jm, owner, release):
+    return jm.start_job(
+        release.wait,
+        origin={"deliver": "origin", "user_id": owner, "task": "research repo"},
+    )
+
+
+def test_list_detail_and_cancel_own_jobs():
+    jm = BackgroundJobManager()
+    release = threading.Event()
+    try:
+        mine = _start(jm, "alice", release)
+        theirs = _start(jm, "bob", release)
+        runner = BackgroundRunner()
+
+        listing = handle_tasks_command("alice", None, runner=runner, job_manager=jm)
+        assert mine in listing
+        assert "research repo" in listing
+        assert theirs not in listing
+
+        detail = handle_tasks_command("alice", mine, runner=runner, job_manager=jm)
+        assert mine in detail
+
+        assert "not found" in handle_tasks_command(
+            "alice", theirs, runner=runner, job_manager=jm
+        )
+        assert "not found" in handle_tasks_command(
+            "alice", f"cancel {theirs}", runner=runner, job_manager=jm
+        )
+
+        out = handle_tasks_command("alice", f"cancel {mine}", runner=runner, job_manager=jm)
+        assert out.startswith("✅")
+        assert jm.get_job_info(mine).status.value == "cancelled"
+        assert jm.get_job_info(theirs).status.value == "running"
+    finally:
+        release.set()
+        jm.shutdown()
+
+
+def test_ownerless_jobs_hidden():
+    jm = BackgroundJobManager()
+    try:
+        job = jm.start_job(lambda: "x")
+        jm.get_result(job, timeout=5)
+        out = handle_tasks_command("alice", None, runner=BackgroundRunner(), job_manager=jm)
+        assert out == "💤 No background tasks."
+    finally:
+        jm.shutdown()
+
+
+def test_newest_jobs_listed_first_when_over_limit():
+    jm = BackgroundJobManager()
+    try:
+        ids = []
+        for _ in range(25):
+            job = jm.start_job(
+                lambda: "x", origin={"deliver": "origin", "user_id": "alice"}
+            )
+            jm.get_result(job, timeout=5)
+            ids.append(job)
+        out = handle_tasks_command("alice", None, runner=BackgroundRunner(), job_manager=jm)
+        assert ids[-1] in out
+        assert ids[0] not in out
+    finally:
+        jm.shutdown()