refactor(OTB-70): split the event bridge into transport, routing, session runtime state and Telegram delivery

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: grinev

src/app/managers/summary-aggregation-manager.ts

@@ -101,7 +101,7 @@ type ToolFileCallback = (fileInfo: ToolFileInfo) => void;
 
 type QuestionCallback = (questions: Question[], requestID: string, sessionId: string) => void;
 
-type QuestionErrorCallback = () => void;
+type QuestionErrorCallback = (sessionId: string) => void;
 
 type ThinkingCallback = (update: ThinkingUpdate) => void;
 
@@ -115,9 +115,9 @@ export interface TokensInfo {
   cacheWrite: number;
 }
 
-type TokensCallback = (tokens: TokensInfo, isCompleted: boolean) => void;
+type TokensCallback = (sessionId: string, tokens: TokensInfo, isCompleted: boolean) => void;
 
-type CostCallback = (cost: number) => void;
+type CostCallback = (sessionId: string, cost: number) => void;
 
 export type SubagentStatus = "pending" | "running" | "completed" | "error";
 
@@ -169,7 +169,7 @@ type PermissionRepliedCallback = (sessionId: string, requestID: string) => void
 
 type SessionDiffCallback = (sessionId: string, diffs: FileChange[]) => void;
 
-type FileChangeCallback = (change: FileChange) => void;
+type FileChangeCallback = (sessionId: string, change: FileChange) => void;
 
 type ClearedCallback = () => void;
 
@@ -599,6 +599,23 @@ export class SummaryAggregator {
     return this.isTrackedChildSession(sessionId);
   }
 
+  /**
+   * The root session a tracked session belongs to, found by walking its parents;
+   * an untracked session is its own root.
+   */
+  getRootSessionId(sessionId: string): string {
+    const visited = new Set<string>();
+    let rootSessionId = sessionId;
+    let parentSessionId = this.trackedSessionParents.get(rootSessionId);
+    while (parentSessionId && !visited.has(parentSessionId)) {
+      visited.add(rootSessionId);
+      rootSessionId = parentSessionId;
+      parentSessionId = this.trackedSessionParents.get(rootSessionId);
+    }
+
+    return rootSessionId;
+  }
+
   private getQueue(map: Map<string, string[]>, parentSessionId: string): string[] {
     const existing = map.get(parentSessionId);
     if (existing) {
@@ -1237,7 +1254,7 @@ export class SummaryAggregator {
           `[Aggregator] Tokens: input=${tokens.input}, output=${tokens.output}, reasoning=${tokens.reasoning}, cacheRead=${tokens.cacheRead}, cacheWrite=${tokens.cacheWrite}, completed=${isCompleted}`,
         );
         // Call synchronously so keyboardManager is updated before onComplete sends the reply
-        this.onTokensCallback(tokens, isCompleted);
+        this.onTokensCallback(info.sessionID, tokens, isCompleted);
       }
 
       if (isCompleted) {
@@ -1261,7 +1278,7 @@ export class SummaryAggregator {
         // Extract and report cost
         if (this.onCostCallback && assistantInfo.cost !== undefined) {
           logger.debug(`[Aggregator] Cost: $${assistantInfo.cost.toFixed(2)}`);
-          this.onCostCallback(assistantInfo.cost);
+          this.onCostCallback(info.sessionID, assistantInfo.cost);
         }
 
         if (this.onCompleteCallback && finalText.length > 0) {
@@ -1463,8 +1480,9 @@ export class SummaryAggregator {
             `[Aggregator] Question tool failed with error, clearing active poll. callID=${part.callID}`,
           );
           if (this.onQuestionErrorCallback) {
+            const sessionId = part.sessionID;
             setImmediate(() => {
-              this.onQuestionErrorCallback!();
+              this.onQuestionErrorCallback!(sessionId);
             });
           }
           return;
@@ -1524,7 +1542,7 @@ export class SummaryAggregator {
           }
 
           if (preparedFileContext.fileChange && this.onFileChangeCallback) {
-            this.onFileChangeCallback(preparedFileContext.fileChange);
+            this.onFileChangeCallback(part.sessionID, preparedFileContext.fileChange);
           }
         }
       }

src/app/services/event-router.ts

@@ -0,0 +1,98 @@
+import type { Event } from "@opencode-ai/sdk/v2";
+import { config } from "../../config.js";
+import type { AppContainer } from "../bootstrap/app-container.js";
+import type { EventEnvelope } from "../../opencode/events.js";
+import { safeBackgroundTask } from "../../utils/safe-background-task.js";
+import { markAttachedSessionBusy } from "./attach-service.js";
+import { reconcileBusyState } from "./busy-reconciliation-service.js";
+import { ingestSessionInfoForCache } from "./session-cache-service.js";
+
+export type EventRouterDeps = Pick<
+  AppContainer,
+  | "assistantRunState"
+  | "attachManager"
+  | "backgroundSessionTracker"
+  | "foregroundSessionState"
+  | "scheduledTaskRuntime"
+  | "summaryAggregator"
+>;
+
+export interface EventRouterOptions {
+  /** The directory the subscription was opened for. */
+  directory: string;
+  deps: EventRouterDeps;
+  isForegroundSession: (sessionId: string) => boolean;
+}
+
+/** The session an event belongs to, read from the event itself. */
+function getEventSessionId(event: Event): string | null {
+  const properties = event.properties as {
+    sessionID?: string;
+    info?: { sessionID?: string };
+    part?: { sessionID?: string };
+  };
+
+  return properties.sessionID || properties.info?.sessionID || properties.part?.sessionID || null;
+}
+
+function shouldMarkAttachedBusyFromEvent(event: Event): boolean {
+  switch (event.type) {
+    case "session.status":
+      return (event.properties as { status?: { type?: string } }).status?.type === "busy";
+    case "message.updated": {
+      const info = (event.properties as { info?: { role?: string; time?: { completed?: number } } })
+        .info;
+      return info?.role === "assistant" && !info.time?.completed;
+    }
+    case "message.part.updated":
+    case "message.part.delta":
+    case "question.asked":
+    case "permission.asked":
+      return true;
+    default:
+      return false;
+  }
+}
+
+/**
+ * Routes events of one subscription. Session identity always comes from the
+ * event; events without one only do subscription-wide work.
+ */
+export function createEventRouter({
+  directory,
+  deps,
+  isForegroundSession,
+}: EventRouterOptions): (envelope: EventEnvelope) => void {
+  return ({ event }) => {
+    // The SDK event union does not list the heartbeat the server sends.
+    if ((event as { type: string }).type === "server.heartbeat") {
+      // A heartbeat is a liveness signal of the subscription, so the check runs
+      // for the directory the subscription was opened with.
+      void reconcileBusyState(directory, deps);
+    }
+
+    const sessionId = getEventSessionId(event);
+    const attached = deps.attachManager.getSnapshot();
+    if (attached && sessionId === attached.sessionId && shouldMarkAttachedBusyFromEvent(event)) {
+      void markAttachedSessionBusy(attached.sessionId, deps);
+    }
+
+    if (event.type === "session.created" || event.type === "session.updated") {
+      const info = event.properties.info;
+
+      if (info?.directory) {
+        safeBackgroundTask({
+          taskName: `session.cache.${event.type}`,
+          task: () => ingestSessionInfoForCache(info),
+        });
+      }
+    }
+
+    if (config.bot.trackBackgroundSessions) {
+      const foregroundSessionId = sessionId && isForegroundSession(sessionId) ? sessionId : null;
+      deps.backgroundSessionTracker.processEvent(event, foregroundSessionId);
+    }
+
+    deps.summaryAggregator.processEvent(event);
+  };
+}

src/bot/events/background-notice-delivery.ts

@@ -0,0 +1,49 @@
+import { t } from "../../i18n/index.js";
+import type { BackgroundSessionNotification } from "../../app/managers/background-session-manager.js";
+import { buildBackgroundSessionOpenKeyboard } from "../menus/session-selection-menu.js";
+import type { SessionTargetPolicy, TelegramEventDelivery } from "./telegram-event-delivery.js";
+
+function formatShortSessionId(sessionId: string): string {
+  return sessionId.length <= 8 ? sessionId : sessionId.slice(0, 8);
+}
+
+function getBackgroundSessionLabel(notification: BackgroundSessionNotification): string {
+  const title = notification.sessionTitle?.trim();
+  if (title) {
+    return title;
+  }
+
+  return t("background.session_fallback", {
+    id: formatShortSessionId(notification.sessionId),
+  });
+}
+
+function formatBackgroundSessionNotification(notification: BackgroundSessionNotification): string {
+  const session = getBackgroundSessionLabel(notification);
+
+  switch (notification.kind) {
+    case "assistant_response":
+      return t("background.assistant_response", { session });
+    case "question_asked":
+      return t("background.question_asked", { session });
+    case "permission_asked":
+      return t("background.permission_asked", { session });
+  }
+}
+
+/** Sends a background-session notice to the destination of its own session. */
+export function createBackgroundNoticeDelivery(
+  policy: Pick<SessionTargetPolicy, "getDestination">,
+  delivery: TelegramEventDelivery,
+): (notification: BackgroundSessionNotification) => Promise<void> {
+  return async (notification) => {
+    const destination = policy.getDestination(notification.sessionId);
+    if (!destination) {
+      return;
+    }
+
+    await delivery.sendText(destination, formatBackgroundSessionNotification(notification), {
+      reply_markup: buildBackgroundSessionOpenKeyboard(notification.sessionId, notification.kind),
+    });
+  };
+}