Fix push notifications for account runtimes

docs/services.md

@@ -11,6 +11,7 @@ This is an ownership map, not an API dump. Read the owning implementation and te
 | `services/reddit-account-runtime.ts` | Immutable account runtime/factory/manager lifecycle | Same-account operations serialize; session refresh replaces one runtime. |
 | `services/reddit-user-messenger.ts` | Account selection for explicit API/MCP messages | Prefer the newest existing conversation owner; otherwise choose the authenticated running account with the fewest hourly sends. |
 | `services/reddit-account-runtime-builder.ts` | Concrete account dependency graph and account loop | Account-specific browser, Matrix, repos, limiter, automation, chat, and media live here. |
+| `services/account-runtime-event-wiring.ts` | Per-account chat-to-realtime event bridge | Every account runtime publishes reply, media, and sync events into the shared dashboard/push stream; sync IDs remain account-scoped. |
 | `services/reddit-accounts.ts` | Account import/activation orchestration | Verification must bind Reddit cookies and Matrix identity before activation. |
 | `services/account-auth-context.ts` | Current verified identity snapshot | Account work pauses unless both sides authenticate. |
 | `services/session-bound-runtime.ts` | Legacy auth-bound background lifecycle | Prevents background work while logged out. |

src/bootstrap.ts

@@ -373,6 +373,7 @@ export async function bootstrap(
       logger,
       db,
       getBotMode: () => webServer?.getMode() ?? 'running',
+      events: broadcaster.dashboardEvents,
     }),
   );
   const redditAccounts = new RedditAccountsService(

src/services/account-runtime-event-wiring.ts

@@ -0,0 +1,82 @@
+import {
+  type ChatService,
+  type SyncCompletion,
+  type SyncProgress,
+} from './chat/index.js';
+
+/** Events emitted by one account-owned chat service into the shared UI/push hub. */
+export interface AccountRuntimeEventReporter {
+  conversation(input: { id: string; state: 'replied'; unread: number }): unknown;
+  reply(input: { conversationId: string; id: string; username: string; body?: string }): unknown;
+  mediaReceived(input: { conversationId: string; username: string; count: number }): unknown;
+  sync(input:
+    | { phase: 'started'; id: string; kind: 'delta' | 'full' }
+    | { phase: 'progress'; id: string; kind: 'delta' | 'full'; current: number; total?: number; label?: string }
+    | { phase: 'completed'; id: string; kind: 'delta' | 'full'; processed: number }
+    | { phase: 'failed'; id: string; kind: 'delta' | 'full'; error: string }
+  ): unknown;
+}
+
+type AccountChatEventSource = Pick<
+  ChatService,
+  'setProgressListener' | 'setNewReplyListener' | 'setMediaReceivedListener'
+>;
+
+/** Connects one account-owned ChatService to the process-wide event stream. */
+export function wireAccountChatEvents(
+  chatService: AccountChatEventSource,
+  events: AccountRuntimeEventReporter,
+  accountId: string,
+): void {
+  let activeSync: SyncProgress | null = null;
+
+  chatService.setProgressListener((progress, completion) => {
+    if (progress) {
+      const id = accountSyncId(accountId, progress.startedAt);
+      if (progress.phase === 'starting') {
+        events.sync({ phase: 'started', id, kind: progress.kind });
+      } else {
+        events.sync({
+          phase: 'progress',
+          id,
+          kind: progress.kind,
+          current: progress.current,
+          total: progress.total,
+          label: progress.label ?? progress.phase,
+        });
+      }
+      activeSync = progress;
+      return;
+    }
+
+    if (!activeSync) return;
+    reportSyncCompletion(events, accountId, activeSync, completion);
+    activeSync = null;
+  });
+
+  chatService.setMediaReceivedListener(({ conversationId, username, count }) => {
+    events.mediaReceived({ conversationId, username, count });
+  });
+  chatService.setNewReplyListener(({ conversationId, replyId, username, replySnippet }) => {
+    events.conversation({ id: conversationId, state: 'replied', unread: 1 });
+    events.reply({ conversationId, id: replyId, username, body: replySnippet });
+  });
+}
+
+function reportSyncCompletion(
+  events: AccountRuntimeEventReporter,
+  accountId: string,
+  progress: SyncProgress,
+  completion?: SyncCompletion,
+): void {
+  const id = accountSyncId(accountId, progress.startedAt);
+  if (completion?.status === 'failed') {
+    events.sync({ phase: 'failed', id, kind: progress.kind, error: completion.error });
+    return;
+  }
+  events.sync({ phase: 'completed', id, kind: progress.kind, processed: progress.current });
+}
+
+function accountSyncId(accountId: string, startedAt: number): string {
+  return `chat-${accountId}-${startedAt}`;
+}

src/services/reddit-account-runtime-builder.ts

@@ -66,12 +66,17 @@ import { AccountGrowthRepository } from '../database/growth-repository.js';
 import { AccountBrowsingSimulationRepository } from '../database/browsing-simulation-repository.js';
 import { AccountBrowsingSimulationService } from './account-browsing-simulation.js';
 import { allowsConversationSync, allowsSending, type BotMode } from './bot-mode.js';
+import {
+  wireAccountChatEvents,
+  type AccountRuntimeEventReporter,
+} from './account-runtime-event-wiring.js';
 
 export interface ConcreteRuntimeFactoryOptions {
   config: Config;
   logger: Logger;
   db: Database.Database;
   getBotMode?: () => BotMode;
+  events?: AccountRuntimeEventReporter;
 }
 
 export function materializePreparedPostForAccount(
@@ -106,7 +111,7 @@ export function createConcreteRedditAccountRuntimeFactory(
 }
 
 async function buildRuntime(
-  { config, logger: rootLogger, db, getBotMode = () => 'running' }: ConcreteRuntimeFactoryOptions,
+  { config, logger: rootLogger, db, getBotMode = () => 'running', events }: ConcreteRuntimeFactoryOptions,
   session: VerifiedAccountSession,
   recipientDeduplicator: RecipientDeduplicator,
   redditRequestCoordinator: RedditRequestCoordinator,
@@ -228,6 +233,7 @@ async function buildRuntime(
       tokenRefresher, usernameResolver, chatCache,
       messagesRepo, redditUsersRepo, accountAuth, conversationOutcomesRepo, conversationStateRepo,
     );
+    if (events) wireAccountChatEvents(chatService, events, session.accountId);
     await chatService.loadCache();
     const mediaSyncService = new MediaSyncService(chatService, logger, accountConfig.media.syncIntervalMs, accountAuth);
     const recipientAccountStatusRepo = new RedditRecipientAccountStatusRepository(db);

tests/account-runtime-event-wiring.test.ts

@@ -0,0 +1,87 @@
+import { describe, expect, it, vi } from 'vitest';
+import {
+  wireAccountChatEvents,
+  type AccountRuntimeEventReporter,
+} from '../src/services/account-runtime-event-wiring.js';
+import type {
+  MediaReceivedListener,
+  NewReplyListener,
+  SyncProgressListener,
+} from '../src/services/chat/index.js';
+
+function harness() {
+  let progressListener: SyncProgressListener | null = null;
+  let replyListener: NewReplyListener | null = null;
+  let mediaListener: MediaReceivedListener | null = null;
+  const chatService = {
+    setProgressListener: vi.fn((listener: SyncProgressListener | null) => { progressListener = listener; }),
+    setNewReplyListener: vi.fn((listener: NewReplyListener | null) => { replyListener = listener; }),
+    setMediaReceivedListener: vi.fn((listener: MediaReceivedListener | null) => { mediaListener = listener; }),
+  };
+  const events = {
+    conversation: vi.fn(),
+    reply: vi.fn(),
+    mediaReceived: vi.fn(),
+    sync: vi.fn(),
+  } satisfies AccountRuntimeEventReporter;
+
+  wireAccountChatEvents(chatService, events, '@t2_account:reddit.com');
+  return {
+    events,
+    progress: (value: Parameters<SyncProgressListener>[0], completion?: Parameters<SyncProgressListener>[1]) => {
+      if (!progressListener) throw new Error('Progress listener was not installed');
+      progressListener(value, completion);
+    },
+    reply: (value: Parameters<NewReplyListener>[0]) => {
+      if (!replyListener) throw new Error('Reply listener was not installed');
+      replyListener(value);
+    },
+    media: (value: Parameters<MediaReceivedListener>[0]) => {
+      if (!mediaListener) throw new Error('Media listener was not installed');
+      mediaListener(value);
+    },
+  };
+}
+
+describe('wireAccountChatEvents', () => {
+  it('publishes replies and media received by an account-owned chat service', () => {
+    const fixture = harness();
+
+    fixture.reply({
+      conversationId: '!room:reddit.com', replyId: 'event-1', username: 'alice', replySnippet: 'hello',
+    });
+    fixture.media({ conversationId: '!room:reddit.com', username: 'alice', count: 2 });
+
+    expect(fixture.events.conversation).toHaveBeenCalledWith({
+      id: '!room:reddit.com', state: 'replied', unread: 1,
+    });
+    expect(fixture.events.reply).toHaveBeenCalledWith({
+      conversationId: '!room:reddit.com', id: 'event-1', username: 'alice', body: 'hello',
+    });
+    expect(fixture.events.mediaReceived).toHaveBeenCalledWith({
+      conversationId: '!room:reddit.com', username: 'alice', count: 2,
+    });
+  });
+
+  it('publishes account-scoped sync progress and failures', () => {
+    const fixture = harness();
+    const progress = {
+      kind: 'delta', phase: 'starting', current: 0, total: 4, startedAt: 123,
+    } as const;
+
+    fixture.progress(progress);
+    fixture.progress({ ...progress, phase: 'listing-rooms', current: 2, label: 'Loading rooms' });
+    fixture.progress(null, { status: 'failed', error: 'Matrix unavailable' });
+
+    expect(fixture.events.sync).toHaveBeenNthCalledWith(1, {
+      phase: 'started', id: 'chat-@t2_account:reddit.com-123', kind: 'delta',
+    });
+    expect(fixture.events.sync).toHaveBeenNthCalledWith(2, {
+      phase: 'progress', id: 'chat-@t2_account:reddit.com-123', kind: 'delta',
+      current: 2, total: 4, label: 'Loading rooms',
+    });
+    expect(fixture.events.sync).toHaveBeenNthCalledWith(3, {
+      phase: 'failed', id: 'chat-@t2_account:reddit.com-123', kind: 'delta', error: 'Matrix unavailable',
+    });
+  });
+});