Services and the Bus

Discovery: nesting is the wiring

A component finds its services element by walking up its ancestors — and checking each ancestor's own children, because both nesting directions are legitimate:

<!-- the page owns the services -->
<machvive-chat-syncopation-services>
  <machvive-chat-syncopation-canvas></machvive-chat-syncopation-canvas>
</machvive-chat-syncopation-services>

<!-- the container made its own -->
<machvive-chat-syncopation>
  <machvive-chat-syncopation-services></machvive-chat-syncopation-services>
  <machvive-chat-syncopation-canvas></machvive-chat-syncopation-canvas>
</machvive-chat-syncopation>

The walk crosses shadow boundaries via the host, so a component living inside someone else's shadow root still works.

Why not a module-level singleton? Because two independent chat surfaces on one page must not share a conversation, and a global registry makes that impossible to express. Position in the document is already the most natural way to say "these belong together", so we use it and add nothing.

When a component receives nothing

This is the failure mode, and it is almost always the same cause: the component is not inside — or beside — the services element it is meant to use.

Drop in <machvive-chat-syncopation-inspector>. If it reports "No <machvive-chat-syncopation-services> found", that is your answer. If it shows traffic but your component is quiet, your component is bound to a different services element than the one producing the traffic.

Waiting for services

Upgrade order is not something a component can rely on — innerHTML, a framework's hydration and document.write each order it differently. So components await discovery rather than assuming it:

import { whenServices } from '@machfivetechchicago/machvive-chat-syncopation-ai/services';

const services = await whenServices(myElement);   // null after ~2s
if (!services) return;                            // render an explanation

findServices(node) is the synchronous version, for when you know the tree is built.

The bus

const off = services.bus.on('record:added', (record) => { /* … */ });
off();                                     // unsubscribe

services.bus.once('services:ready', ({ config }) => { /* … */ });
services.bus.emit('my-app:thing', { any: 'payload' });
services.bus.topics;                       // what has been subscribed to

Topics

Topic	Payload	When
services:ready	{ config }	Services booted
record:added	the record	A turn was added
record:appended	{ id, chunk, record }	A streaming chunk arrived
record:updated	the record	Status or meta changed
conversation:cleared	{ id }	The transcript was emptied
daemon:idle	{ id }	Generation finished, failed or was stopped
daemon:error	{ id, error, name }	A turn failed (error is a string — plain data, so recordings survive JSON)
prompt:fill	{ text, send }	Something wants text in the composer
config:changed	{ config, previous? }	Services configuration changed at runtime

Emit your own freely; the namespace is yours. A prefix (my-app:) keeps them legible in the inspector.

The wildcard

services.bus.on('*', ({ topic, payload }) => console.log(topic, payload));

Every topic, including ones that did not exist when the subscriber was written. This is how the inspector works, and it is worth knowing about for logging and analytics.

Recording and replay

The wildcard is also what makes the bus recordable. The services element owns a Recorder that subscribes to '*' and captures every emission as plain data — { seq, at, topic, payload } — snapshotted at emit time.

services.recorder.start();
await services.send('hello');
services.recorder.stop();

const json = services.recorder.export();   // mcs-recording@1, JSON-serializable
recorder.import(json);                      // here or in another page entirely
await recorder.replay();                    // re-emit instantly
await recorder.replay({ pace: 'realtime' }); // or with the original timing

Two invariants keep a recording honest:

• Payloads are snapshotted, so a streaming turn that keeps mutating after record:added does not rewrite history retroactively.
• Every payload is JSON-safe — including daemon:error, which carries the error message rather than the live Error. A recording that cannot survive JSON.stringify is not a recording, it is a memory leak with good intentions.

Replay re-emits the events onto the bus, so components re-react exactly as they did live. That is what makes a recording a test fixture, a support artifact, or a demo script without any extra machinery.

Always unsubscribe

on returns an unsubscribe function. Call it in disconnectedCallback:

class MySurface extends HTMLElement {
  #off = [];

  async connectedCallback() {
    const services = await whenServices(this);
    if (!services || !this.isConnected) return;
    this.#off = [services.bus.on('record:added', (r) => this.render(r))];
  }

  disconnectedCallback() {
    for (const off of this.#off) off();
    this.#off = [];
  }
}

A detached component that keeps rendering is a leak that also keeps the entire conversation alive in memory. Every component in this package does exactly the above, and a test asserts it.

Two properties worth relying on

A subscriber that throws cannot silence its siblings. emit catches per handler and warns. A chat surface has many independent listeners and one bad render must not take the rest down.

PubSub is not an EventTarget. Deliberately. EventTarget binds the bus to whichever realm supplied the global, and dispatching an event constructed in another realm throws — which surfaces as a component failure rather than a plumbing one. The hand-rolled listener set is realm-free and works in a worker or in plain Node. (We learned this the expensive way in the sibling package, where a successful tool call was logged as an error because of exactly this mismatch.)

The conversation

Conversation is the only thing allowed to mutate the record list, and it announces every change.

const { conversation } = services;

conversation.records;                      // a copy, always
conversation.length;
conversation.last;

conversation.add({ role: 'system', text: 'Order #1234 is open in another tab.' });
conversation.append(id, ' more text');     // streaming
conversation.update(id, { status: 'complete' });
conversation.clear();
conversation.toJSON();                     // { id, records }

records returns a copy because handing out the live array lets any component corrupt the transcript — and a corrupted transcript is not recoverable, since it is the record of what was actually said.

Turns past max-turns are trimmed oldest-first. Recent turns are the ones a reader needs.

Injecting context

add accepts any role, which is the hook for surfacing application state in the conversation itself:

cart.on('change', (cart) => {
  services.conversation.add({
    role: 'system',
    text: `Cart now contains ${cart.items.length} items.`
  });
});

System and tool turns render muted and monospaced, so they read as machinery rather than as something a person said.

The daemon

services.daemon.busy;        // generating?
services.daemon.stop();      // interrupt; the partial turn is kept
await services.daemon.send('Hello');

Three behaviours to know:

A failed turn is visible. An exception in a transport becomes a record with status: 'error' and META.ERROR set, plus a daemon:error event. Silently dropping it leaves the user staring at a prompt that appears to have done nothing.

stop() keeps the partial text. The user already read it; deleting it is disorienting.

Concurrent send is refused and returns null. Interleaving two streams into one transcript produces output no reader can attribute to anything.

Next: Writing a Transport.