refactor(runtime-host): consolidate host change broadcasts

packages/runtime-host/src/__tests__/host-change-feed.test.ts

@@ -0,0 +1,69 @@
+import assert from 'node:assert/strict';
+import { test } from 'node:test';
+import { HostChangeFeed } from '../server/host-change-feed.js';
+
+test('routes each change kind only to subscribed connections', () => {
+  const feed = new HostChangeFeed();
+  const configuration: unknown[] = [];
+  const project: unknown[] = [];
+  const all: unknown[] = [];
+  feed.attachConnection(
+    'configuration',
+    { configuration: true },
+    { send: async (frame) => void configuration.push(frame) },
+  );
+  feed.attachConnection(
+    'project',
+    { projectCatalog: true },
+    { send: async (frame) => void project.push(frame) },
+  );
+  feed.attachConnection(
+    'all',
+    { configuration: true, projectCatalog: true, sessionCatalog: true, scheduledTask: true },
+    { send: async (frame) => void all.push(frame) },
+  );
+
+  feed.publishConfiguration();
+  feed.publishProjectCatalog();
+  feed.publishSessionCatalog('session-1');
+  feed.publishScheduledTask(7, 'updated', 'task-1');
+
+  assert.deepEqual(
+    configuration.map((frame) => (frame as { kind: string }).kind),
+    ['configuration.changed'],
+  );
+  assert.deepEqual(
+    project.map((frame) => (frame as { kind: string }).kind),
+    ['project.catalog.changed'],
+  );
+  assert.equal(all.length, 4);
+});
+
+test('keeps catalog revisions independent and removes failed subscriptions', async () => {
+  const feed = new HostChangeFeed();
+  const frames: unknown[] = [];
+  feed.attachConnection(
+    'working',
+    { projectCatalog: true, sessionCatalog: true },
+    { send: async (frame) => void frames.push(frame) },
+  );
+  feed.attachConnection(
+    'failed',
+    { projectCatalog: true },
+    {
+      send: async () => {
+        throw new Error('closed');
+      },
+    },
+  );
+
+  feed.publishProjectCatalog();
+  feed.publishSessionCatalog('session-1');
+  await Promise.resolve();
+  feed.publishProjectCatalog();
+
+  assert.deepEqual(
+    frames.map((frame) => (frame as { kind: string; revision: number }).revision),
+    [1, 1, 2],
+  );
+});

packages/runtime-host/src/__tests__/host-kernel.test.ts

@@ -65,8 +65,7 @@ import {
 } from '../server/candidate.js';
 import type { RuntimeHostCompositionSource } from '../server/host-composition.js';
 import { createUnavailableDomainOperationHandlers } from '../server/operation-dispatcher.js';
-import { HostConfigurationChangeService } from '../server/configuration-change-service.js';
-import { HostSessionCatalogChangeService } from '../server/session-catalog-change-service.js';
+import { HostChangeFeed } from '../server/host-change-feed.js';
 import { FramedTransport, RuntimeHostTransportError } from '../transport/framed-transport.js';
 import {
   prepareStorageRootControlDirectory,
@@ -1880,8 +1879,7 @@ describe('non-serving Runtime Host kernel', () => {
       const owner = await tryAcquireInteractiveRootOwner(capability);
       assert.ok(owner);
       if (!owner) return;
-      const configurationChanges = new HostConfigurationChangeService();
-      const sessionCatalogChanges = new HostSessionCatalogChangeService();
+      const hostChanges = new HostChangeFeed();
       let releaseFactory!: () => void;
       let markFactoryEntered!: () => void;
       const factoryEntered = new Promise<void>((resolve) => {
@@ -1898,8 +1896,7 @@ describe('non-serving Runtime Host kernel', () => {
           await factoryReleased;
           return {
             handlers: createUnavailableDomainOperationHandlers(),
-            configurationChanges,
-            sessionCatalogChanges,
+            hostChanges,
             beginDrain() {},
             async recover() {},
             async close() {},
@@ -1927,8 +1924,8 @@ describe('non-serving Runtime Host kernel', () => {
         });
         releaseFactory();
         host = await hostTask;
-        configurationChanges.publish();
-        sessionCatalogChanges.publish('session-1');
+        hostChanges.publishConfiguration();
+        hostChanges.publishSessionCatalog('session-1');
         assert.equal(
           await withTimeout(observed, 1_000, 'Client did not receive configuration change'),
           1,

packages/runtime-host/src/__tests__/project-catalog-coordinator.test.ts

@@ -5,10 +5,9 @@ import { join } from 'node:path';
 import { test } from 'node:test';
 import { createProjectCatalog, createSessionStore } from '@maka/storage';
 import type { ConnectionContext } from '../server/operation-dispatcher.js';
-import { HostProjectCatalogChangeService } from '../server/project-catalog-change-service.js';
+import { HostChangeFeed } from '../server/host-change-feed.js';
 import { HostProjectCatalogCoordinator } from '../server/project-catalog-coordinator.js';
 import { HostProjectMembershipGate } from '../server/project-membership-gate.js';
-import { HostSessionCatalogChangeService } from '../server/session-catalog-change-service.js';
 
 test('Host Project Catalog relink merges identities and reassigns every affected Session', async () => {
   const base = await mkdtemp(join(tmpdir(), 'maka-host-project-catalog-'));
@@ -24,30 +23,36 @@ test('Host Project Catalog relink merges identities and reassigns every affected
     })(),
   });
   const sessions = createSessionStore(storageRoot);
-  const projectChanges = new HostProjectCatalogChangeService();
-  const sessionChanges = new HostSessionCatalogChangeService();
+  const hostChanges = new HostChangeFeed();
   const projectFrames: unknown[] = [];
   const sessionFrames: unknown[] = [];
-  projectChanges.attachConnection('desktop', {
-    send: async (frame) => {
-      projectFrames.push(frame);
+  hostChanges.attachConnection(
+    'desktop',
+    {
+      projectCatalog: true,
+      sessionCatalog: true,
     },
-  });
-  projectChanges.attachConnection('tui', {
-    send: async (frame) => {
-      projectFrames.push(frame);
+    {
+      send: async (frame) => {
+        if (frame.kind === 'project.catalog.changed') projectFrames.push(frame);
+        else sessionFrames.push(frame);
+      },
     },
-  });
-  sessionChanges.attachConnection('desktop', {
-    send: async (frame) => {
-      sessionFrames.push(frame);
+  );
+  hostChanges.attachConnection(
+    'tui',
+    { projectCatalog: true },
+    {
+      send: async (frame) => {
+        projectFrames.push(frame);
+      },
     },
-  });
+  );
   const membership = new HostProjectMembershipGate();
   const coordinator = new HostProjectCatalogCoordinator(
     catalog,
-    projectChanges,
-    sessionChanges,
+    { publish: () => hostChanges.publishProjectCatalog() },
+    { publish: (sessionId: string) => hostChanges.publishSessionCatalog(sessionId) },
     membership,
     () => assert.fail('ordinary project mutations must not drain the Host'),
   );
@@ -129,10 +134,11 @@ test('directory resolution failures cannot enter the unknown-commit drain path',
   const base = await mkdtemp(join(tmpdir(), 'maka-host-project-directory-failure-'));
   const catalog = createProjectCatalog(join(base, 'storage'));
   let drains = 0;
+  const hostChanges = new HostChangeFeed();
   const coordinator = new HostProjectCatalogCoordinator(
     catalog,
-    new HostProjectCatalogChangeService(),
-    new HostSessionCatalogChangeService(),
+    { publish: () => hostChanges.publishProjectCatalog() },
+    { publish: (sessionId: string) => hostChanges.publishSessionCatalog(sessionId) },
     new HostProjectMembershipGate(),
     () => {
       drains += 1;