fix(teams): acknowledge a Graph notification once its job is persisted, not after processing

The msgraph webhook awaits its notification scheduler before answering 202 (so an admission
failure can answer 503). The Teams pipeline scheduler awaited run_notification(), which runs the
whole job inline: Graph meeting fetch, transcript download/transcription and summarization. The
202 therefore waited behind minutes of work, against Graph's guidance to persist and respond
promptly, and slow runs invite Graph retries.

The scheduler now only accepts in-request (create_job_from_notification persists the job and its
dedupe receipt) and runs run_job detached on the adapter's background-task set. Duplicates of a
running or finished job return without starting another run. A failed accept still raises into
the webhook's 503 path.

Review: https://github.com/NousResearch/hermes-agent/pull/106742#pullrequestreview-5463585403
(cherry picked from commit ac7cebfc4189a2fde649f1483e7198a05125583c)

plugins/teams_pipeline/runtime.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import logging
 from typing import Any
 
@@ -87,8 +88,17 @@ def bind_gateway_runtime(gateway: Any) -> bool:
         adapter.set_notification_scheduler(_drop)
         return False
 
+    # Graph wants a prompt 202 once the work is durable: accept (persist the job) inside the request,
+    # run the fetch/transcript/summary pipeline detached. An accept failure still fails the request.
+    background = getattr(adapter, "_background_tasks", set())
+
     async def _schedule(notification: dict[str, Any], event: Any) -> None:
-        await runtime.run_notification(notification)
+        job = runtime.create_job_from_notification(notification)
+        if job.status != "received":  # duplicate of a running or finished job
+            return
+        task = asyncio.create_task(runtime.run_job(job.job_id))
+        background.add(task)
+        task.add_done_callback(background.discard)
     adapter.set_notification_scheduler(_schedule)
     gateway._teams_pipeline_runtime = runtime
     gateway._teams_pipeline_runtime_error = None

tests/plugins/test_teams_pipeline_plugin.py

@@ -81,16 +81,27 @@ async def test_bind_gateway_runtime_attaches_scheduler(monkeypatch, tmp_path):
     class FakeAdapter:
         def __init__(self) -> None:
             self.scheduler = None
+            self._background_tasks = set()
 
         def set_notification_scheduler(self, scheduler) -> None:
             self.scheduler = scheduler
 
     class FakePipeline:
+        """Accept persists the job; processing (Graph fetch, transcript, summary) blocks until released."""
+
         def __init__(self) -> None:
             self.notifications = []
+            self.release = asyncio.Event()
 
-        async def run_notification(self, notification):
-            self.notifications.append(notification)
+        def create_job_from_notification(self, notification):
+            return SimpleNamespace(job_id=f"job-{notification['id']}", status="received")
+
+        async def run_job(self, job_id):
+            await self.release.wait()
+            self.notifications.append({"id": job_id.removeprefix("job-")})
+
+        async def run_notification(self, notification):  # TeamsMeetingPipeline: accept, then run inline
+            return await self.run_job(self.create_job_from_notification(notification).job_id)
 
     adapter = FakeAdapter()
     pipeline = FakePipeline()
@@ -110,7 +121,11 @@ async def test_bind_gateway_runtime_attaches_scheduler(monkeypatch, tmp_path):
     assert callable(adapter.scheduler)
 
     notification = {"id": "notif-1"}
-    await adapter.scheduler(notification, object())
+    # The webhook's 202 waits only for the durable accept, never for the processing pipeline.
+    await asyncio.wait_for(adapter.scheduler(notification, object()), timeout=1)
+    assert pipeline.notifications == []
+    pipeline.release.set()
+    await asyncio.gather(*adapter._background_tasks)
     assert pipeline.notifications == [notification]