feat(desktop): Journal-driven companion bubbles and notification preferences

PLANS.md

@@ -138,3 +138,4 @@ the durable truth after it changes. Git history is the archive.
   in `dev`; remaining work is the TypeScript CLI conversion, logs/Doctor/update
   UX, config check, registry deletion, authenticity-hardened updates, and
   release-gate N-1.
+- [[plans/companion-activity-notifications.md]] — Shared Journal notification preferences and background companion bubbles; Draft PR only, no release authority.

apps/desktop/src/activity-announcements.ts

@@ -0,0 +1,93 @@
+import type { AgentActivitySignal } from './activity-projection.js'
+import type { NotificationInput } from './notification-queue.js'
+function operationId(signal: AgentActivitySignal): string {
+  return signal.operationId ?? (signal.taskId ? `task:${signal.taskId}`
+    : signal.workspaceId && signal.resumeId ? `session:${signal.workspaceId}:${signal.resumeId}` : `event:${signal.revision}`)
+}
+
+
+export interface AnnouncementContext {
+  agent(signal: AgentActivitySignal): string
+  action(signal: AgentActivitySignal): NotificationInput['action']
+  t(key: string, values?: { agent?: string }): string
+  open(context: 'office' | 'inbox' | 'news'): void
+}
+export class ActivityAnnouncements {
+  private announcedThrough: number | null = null
+  private running = new Set<string>()
+  private delivered = new Map<string, number>()
+  private failed = new Map<string, number>()
+  baseline(revision: number) { this.announcedThrough = revision }
+  accept(signals: readonly AgentActivitySignal[], queue: { publish(input: NotificationInput): unknown; withdraw(id: string): void }, context: AnnouncementContext) {
+    if (this.announcedThrough === null) { this.baseline(Math.max(0, ...signals.map(s => s.revision))); return }
+    for (const map of [this.delivered, this.failed]) {
+      while (map.size > 500) map.delete(map.keys().next().value!)
+    }
+    const active = new Set(signals.filter(signal => signal.kind === 'conversation' || signal.kind === 'sonner-test-running')
+      .map(signal => signal.kind === 'conversation' ? `activity:${operationId(signal)}` : `activity:${signal.id}`))
+
+    // Announce in journal order. A polling replay (even >200 facts) never emits twice.
+    for (const signal of [...signals].sort((a, b) => a.revision - b.revision)) {
+      if (signal.revision <= this.announcedThrough) continue
+      this.announcedThrough = signal.revision
+      const operation = operationId(signal)
+      const id = `activity:${operation}`
+      const agent = context.agent(signal)
+      const inspect = context.action(signal)
+      const common = { id, revision: signal.revision, action: inspect }
+      let input: NotificationInput
+      switch (signal.kind) {
+        case 'conversation':
+          if (this.delivered.has(operation) || this.failed.has(operation)) continue
+          input = { ...common, status: 'running', title: agent,
+            description: context.t('activityToast.conversationRunning', { agent }), duration: Infinity }
+          break
+        case 'conversation-failed':
+          this.failed.set(operation, signal.revision)
+          input = { ...common, status: signal.failureKind === 'rejected' ? 'warning' : 'error',
+            title: context.t(signal.failureKind === 'spawn' ? 'activityToast.conversationSpawnFailed'
+              : signal.failureKind === 'rejected' ? 'activityToast.conversationRejected' : 'activityToast.conversationFailed', { agent }),
+            description: signal.detail, duration: signal.failureKind === 'rejected' ? 8_000 : 10_000 }
+          break
+        case 'conversation-completed':
+          if (this.delivered.has(operation) || this.failed.has(operation)) continue
+          input = { ...common, status: 'success', title: context.t('activityToast.conversationCompleted', { agent }), duration: 4_000 }
+          break
+        case 'conversation-interrupted':
+        case 'conversation-paused':
+          input = { ...common, status: 'neutral', title: context.t(signal.kind === 'conversation-paused'
+            ? 'activityToast.conversationPaused' : 'activityToast.conversationInterrupted', { agent }), duration: 4_000 }
+          break
+        case 'inbox': {
+          const related = signal.taskId ? `task:${signal.taskId}` : undefined
+          // If failure is already known, the report is a delivery fact, not a successful execution.
+          const reportsFailure = related && this.failed.has(related)
+          if (related) this.delivered.set(related, signal.revision)
+          input = {
+            ...(related && !reportsFailure ? { id: `activity:${related}` } : {}),
+            group: `inbox:${signal.workspaceId}:${signal.sessionRecordId ?? signal.resumeId ?? signal.agent ?? 'unknown'}`,
+            windowMs: 4_000, revision: signal.revision, status: 'success',
+            title: context.t('activityToast.inboxDelivered', { agent }), description: signal.detail, duration: 6_000,
+            action: { label: context.t('activityToast.viewInbox'), onClick: () => context.open('inbox') },
+          }
+          break
+        }
+        case 'news':
+          input = { status: 'info', title: signal.source ?? context.t('activityToast.newsSource'),
+            description: signal.detail, articleId: signal.newsItemId, image: signal.image,
+            group: `news:${(signal.source ?? '').trim().toLowerCase()}`, windowMs: 4_000,
+            revision: signal.revision, duration: 6_000,
+            action: { label: context.t('activityToast.viewNews'), onClick: () => context.open('news') },
+          }
+          break
+        default:
+          input = { id: `activity:${signal.id}`, revision: signal.revision,
+            status: signal.kind === 'sonner-test-running' ? 'running' : signal.kind === 'sonner-test-success' ? 'success' : 'error',
+            title: signal.detail ?? 'Sonner test', duration: signal.kind === 'sonner-test-running' ? Infinity : signal.kind === 'sonner-test-success' ? 4_000 : 10_000 }
+      }
+      queue.publish(input)
+    }
+    for (const id of this.running) if (!active.has(id)) queue.withdraw(id)
+    this.running = active
+  }
+}

apps/desktop/src/activity-controller.spec.ts

@@ -0,0 +1,90 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import { ActivityController, type ActivityDisplay } from './activity-controller.js'
+import { DEFAULT_ACTIVITY_PREFERENCES as defaults } from './activity-policy.js'
+import type { AgentRuntimeEvent } from './activity-journal-types.js'
+function event(seq: number, type = 'dev.sonner_test', payload: Record<string, unknown> = { testState: 'success', message: 'PRIVATE' }): AgentRuntimeEvent {
+  return { seq, ts: Date.now(), type: type as AgentRuntimeEvent['type'], payload }
+}
+function setup() {
+  let identity: string | null = 'project-a:1', foreground = true, petVisible = true
+  const read = vi.fn().mockResolvedValue({ entries: [], lastSeq: 0 })
+  const sent: Array<{ surface: string; event: ActivityDisplay }> = []
+  const open = vi.fn()
+  const controller = new ActivityController({ identity: () => identity, read }, {
+    foreground: () => foreground, petVisible: () => petVisible, open,
+    send: (surface, event) => sent.push({ surface, event }),
+  }, structuredClone(defaults))
+  return { controller, read, sent, open,
+    setIdentity: (value: string | null) => { identity = value },
+    background: () => { foreground = false; controller.transition() },
+    hidePet: () => { petVisible = false; controller.transition() },
+    page: async (entries: AgentRuntimeEvent[], lastSeq = Math.max(0, ...entries.map(e => e.seq))) => {
+      read.mockResolvedValueOnce({ entries, lastSeq }); await controller.poll()
+    },
+  }
+}
+beforeEach(() => { vi.useFakeTimers(); vi.setSystemTime(new Date('2026-09-30T12:00:00Z')) })
+afterEach(() => { vi.useRealTimers() })
+describe('desktop Journal arbitration', () => {
+  it('silences historical baseline, announces one new revision and keeps replay silent', async () => {
+    const s = setup(); await s.page([event(1)]); expect(s.sent).toEqual([])
+    await s.page([event(2)]); expect(s.sent.filter(e => e.event.type === 'show')).toHaveLength(1)
+    await s.page([event(2)]); expect(s.sent.filter(e => e.event.type === 'show')).toHaveLength(1); s.controller.stop()
+  })
+  it('chooses foreground vs already visible pet, keeps privacy and never replays on transition', async () => {
+    const s = setup(); await s.page([]); await s.page([event(1)])
+    const first = s.sent[0]; expect(first.surface).toBe('main')
+    s.background(); await s.page([event(1)]); expect(s.sent.filter(e => e.event.type === 'show')).toHaveLength(1)
+    await s.page([event(2)]); const pet = s.sent.at(-1)!; expect(pet.surface).toBe('pet')
+    expect(JSON.stringify(pet)).not.toContain('PRIVATE'); expect(pet.event).toMatchObject({ input: { title: 'Work completed' } })
+    s.hidePet(); await s.page([event(3)]); expect(s.sent.filter(e => e.event.type === 'show')).toHaveLength(2); s.controller.stop()
+  })
+  it('rejects stale/invalid actions and routes only active matching surface tokens', async () => {
+    const s = setup(); await s.page([]); s.background(); await s.page([event(1,'news.ingested',{newsItemId: 3,source:'Private source',title:'Private headline'})])
+    const displayId = s.sent[0].event.displayId
+    expect(s.controller.open({url:'https://evil'}, 'pet')).toBe(false)
+    expect(s.controller.open(displayId, 'main')).toBe(false)
+    expect(s.controller.open(displayId,'pet')).toBe(true); expect(s.open).toHaveBeenCalledWith('news')
+    expect(s.controller.open(displayId,'pet')).toBe(false)
+    await s.page([event(2)]); const stale = s.sent.at(-1)!.event.displayId; s.setIdentity('project-b:2')
+    expect(s.controller.open(stale,'pet')).toBe(false); s.controller.stop()
+  })
+  it('discards old-source in-flight responses and baselines the new project', async () => {
+    const s = setup(); await s.page([])
+    let resolve!: (value: unknown) => void
+    s.read.mockReturnValueOnce(new Promise(r => { resolve = r }))
+    const request = s.controller.poll(); s.setIdentity('project-b:2'); s.read.mockResolvedValueOnce({entries:[event(5)],lastSeq:5}); await s.controller.poll()
+    resolve({ entries: [event(1)], lastSeq: 1 }); await request
+    await s.page([event(5)]); expect(s.sent).toEqual([])
+    await s.page([event(6)]); expect(s.sent[0].event.type).toBe('show'); s.controller.stop()
+  })
+  it('bounds reconnect backlog, rejects invalid schema, serializes polling and resets restarted journals', async () => {
+    const s = setup(); await s.page([])
+    s.read.mockRejectedValueOnce(new Error('offline')); await s.controller.poll()
+    await s.page([event(1)],1000); expect(s.sent).toEqual([])
+    await s.page([event(1001)]); expect(s.sent.filter(e => e.event.type === 'show')).toHaveLength(1)
+    s.read.mockResolvedValueOnce({entries:'invalid', lastSeq:1002}); await s.controller.poll()
+    await s.page([event(1)]); expect(s.sent.filter(e => e.event.type === 'show')).toHaveLength(1)
+    await s.page([event(2)]); expect(s.sent.filter(e => e.event.type === 'show')).toHaveLength(2); s.controller.stop()
+  })
+  it('coalesces news, preempts with errors, expires/dismisses without task mutation', async () => {
+    const s = setup(); await s.page([]); s.background()
+    await s.page([event(1,'news.ingested',{newsItemId:1,source:'feed',title:'a'}),event(2,'news.ingested',{newsItemId:2,source:'feed',title:'b'})])
+    expect(s.sent.at(-1)?.event).toMatchObject({ input: { count:2, title:'News updated' } })
+    await s.page([event(3,'dev.sonner_test',{testState:'error',message:'error'})]); expect(s.sent.at(-1)?.event).toMatchObject({input:{status:'error'}})
+    const id=s.sent.at(-1)!.event.displayId; expect(s.controller.dismiss(id,'pet')).toBe(true); expect(s.open).not.toHaveBeenCalled()
+    await vi.advanceTimersByTimeAsync(10_000); s.controller.stop()
+  })
+  it('shares preference filtering, pause and lifecycle updates', async () => {
+    const s = setup(); await s.page([])
+    await s.page([event(1,'dev.sonner_test',{testState:'running'})]); expect(s.sent).toEqual([])
+    s.controller.setPreferences({...defaults, events:{...defaults.events,progress:true}})
+    await s.page([event(2,'runtime.started',{workspaceId:'w',resumeId:'r',taskId:'t',agent:'a',cause:{kind:'conversation',from:{kind:'session'}}})])
+    const id=s.sent.at(-1)!.event.displayId
+    await s.page([event(3,'runtime.stopped',{workspaceId:'w',resumeId:'r',taskId:'t',status:'done'})])
+    expect(s.sent.at(-1)?.event).toMatchObject({ displayId:id, input:{status:'success'} })
+    s.controller.setPreferences({...defaults, pausedUntil:Date.now()+3600000})
+    await s.page([event(4)]); const count=s.sent.filter(e=>e.event.type==='show').length
+    await vi.advanceTimersByTimeAsync(3600001); await s.page([event(5)]); expect(s.sent.filter(e=>e.event.type==='show')).toHaveLength(count+1);s.controller.stop()
+  })
+})