Record objects' HTTP traffic as typeId-keyed cassettes

[andreBurnt]

Record objects' HTTP traffic as typeId-keyed cassettes

Piece 1 of the #11 series, per the order we agreed there: the recorder first, so evidence exists before anything judges it. No judging in this PR, no manifest declarations, no new dependencies. 3 commits, 4 files, +385/-5.

What this adds

Every HTTP request an object makes through HttpClient now leaves a trace: a cassette under cassettes:<typeId> in Storage, written by a CassetteRecorder abject spawned permanent at boot. Restart the object and the evidence is still there, because it is keyed by the durable typeId, not the ephemeral AbjectId.

The seam

I asked in #11 whether you wanted an open exchange event or a scoped subscribe message. Reading the dependents protocol answered it: changed() already IS both. HttpClient emits changed('httpExchange', ...), and only objects that sent addDependent receive it. Anything on the bus may subscribe (your everything-on-the-bus spirit), nothing sees traffic without subscribing (my scoping concern). If you want a different shape, the emission is one method and moves cheaply.

Everything crosses the bus as messages: the subscription (addDependent), the exchanges (events), the persistence (Storage requests), the attribution lookups (Registry requests). HttpClient and the recorder are both workerEligible and may sit on different threads. There is no module state anywhere in the path.

One line lands in core/abject.ts: a protected hasDependents accessor, so an unobserved HttpClient skips building event payloads entirely. Zero work when nobody subscribed.

What the redaction does, and does not, promise

Redaction is stem-based on NAMES, applied before emission inside HttpClient: headers, query params, and JSON/form body fields whose name contains a secret stem (key, token, secret, passw, credential, session, signature, cookie, auth) are replaced with REDACTED before anything crosses the bus. That catches authorization, client_secret, refresh_token, x-amz-security-token, X-Amz-Signature, an access_token in an OAuth response body, a password in a form post - and whatever similarly-named header a generated object invents. A false positive redacts something harmless; a false negative persists a live credential, so the matcher errs toward matching.

What it does NOT do: find a secret stored under an innocent name, or scan free text. It is a name-pattern scrub, not a secret scanner. If you want value-based redaction against what SecretsVault actually holds, that is a conversation I would have before building it - it means the vault's values reach HttpClient for comparison, which has its own threat model. The request handed to the network is untouched either way; only the emitted copy is scrubbed.

Two more honest boundaries: bodies are capped at 64K characters (not bytes) with a truncated flag, and requests that throw before a response exists (timeout, DNS failure, your SSRF guard) produce no cassette - only resolved responses are evidence, including 4xx/5xx. If you want transport-failure evidence too, that is an event-shape question and I would rather design it with you than guess.

Design choices you should check

1. Attribution is registry-verified. The event carries msg.routing.from, and the recorder resolves it through resolveCallerIdentity - your existing anti-spoofing helper - rather than trusting the payload. A caller with no durable typeId is not recorded: a cassette that cannot be tied to a type is evidence about nothing.

2. Eviction is bucketed by endpoint (method + path, query stripped): FIFO cap 5 per bucket, 50 per typeId, and global overflow always comes out of the LARGEST bucket. Your Fitness gate: judge generated objects by evidence they cannot edit #11 bucketing suggestion, landed here where _http volume makes it real - a hot parameterized endpoint cannot evict the one recording of a rarely-hit one, no matter how old that recording is.

3. The recorder never silently gives up. HttpClient discovery retries with capped backoff for as long as the recorder lives, and the subscription does not wait on Storage. While Storage is missing, recording continues in memory; persistence catches up when it appears, merging what the store already held so a restart never clobbers evidence. Both behaviors exist because scripted verification caught the failure modes, not because I thought of them.

4. Discovery is by registered name (discoverDep('HttpClient') / discoverDep('Storage')), the same pattern HealthMonitor and the Abject base class already use. I know your rule about naming specific objects - the sandbox hardcoding you flagged in Fitness gate: judge generated objects by evidence they cannot edit #11 special-cased behavior per name, which this does not. If you want capability-based discovery here instead, say so and I will follow whatever HealthMonitor migrates to.

5. The cassette shape is contributor-authored infrastructure, not generator-authored evidence. Your Fitness gate: judge generated objects by evidence they cannot edit #11 concern was the generator declaring the standards it is judged by. Nothing here is written by a generator: the recorder records what actually crossed the wire, and a judged object cannot influence what gets recorded about it. The shape itself is yours to review like any other code.

6. fetchBase64 does not emit. Base64 image fetches are not replay evidence and would blow the body cap for nothing. Say the word if you want it covered.

Verification

tsc --noEmit clean. Three one-shot scripts (below, not committed - your convention), 16 checks total, run with npx tsx --test <file>:

1. recorder-unit.ts - recorder behaviors against the real bus and real dependents protocol: typeId keying, no-typeId skip, merge-not-clobber on restart, per-bucket eviction, largest-bucket overflow eviction, late-registering HttpClient, recording through a Storage outage.
2. http-emit-unit.ts - emission behaviors with only makeRequest stubbed (your SSRF guard blocks loopback, so no live socket): caller attribution, header/query/body redaction incl. the stem cases above, truncation, all four entry points, zero work when nobody subscribes.
3. recorder-live.ts - end to end with production classes: real Registry, real Storage, real HttpClient, real recorder. Boot-style spawn and register, discovery, subscription, a caller's request, cassette read back from Storage with the secret gone.

lab/recorder-unit.ts

/**
 * One-shot verification for the CassetteRecorder abject (PR-A, abject#11 series).
 * Not committed — repo convention keeps tests out of the tree; this script goes
 * in the PR description. Run: npx tsx --test lab/recorder-unit.ts
 *
 * Real plumbing: a real MessageBus, real Abject subclasses, real dependents
 * protocol. Only Registry and Storage are test doubles (in-memory), speaking
 * the exact message protocol of the real ones.
 */
import { test } from 'node:test';
import assert from 'node:assert/strict';
import { MessageBus } from '../src/runtime/message-bus.js';
import { Abject } from '../src/core/abject.js';
import { request } from '../src/core/message.js';
import type { AbjectId, AbjectMessage, TypeId } from '../src/core/types.js';
import { CassetteRecorder } from '../src/objects/cassette-recorder.js';

const iface = (name: string) => ({
  name,
  description: `test ${name}`,
  version: '0.0.1',
  interface: { id: `test:${name.toLowerCase()}`, name, description: name, methods: [] },
});

/** In-memory Registry double: discover-by-name + lookup(objectId) -> {name, typeId}. */
class FakeRegistry extends Abject {
  private byName = new Map<string, AbjectId>();
  private identities = new Map<AbjectId, { name: string; typeId?: TypeId }>();
  constructor() { super({ manifest: iface('Registry') }); }
  addEntry(id: AbjectId, name: string, typeId?: TypeId): void {
    this.byName.set(name, id);
    this.identities.set(id, { name, typeId });
  }
  protected async onInit(): Promise<void> {
    this.on('discover', (msg: AbjectMessage) => {
      const { name } = msg.payload as { name: string };
      const id = this.byName.get(name);
      return id ? [{ id }] : [];
    });
    this.on('lookup', (msg: AbjectMessage) => {
      const { objectId } = msg.payload as { objectId: AbjectId };
      return this.identities.get(objectId) ?? null;
    });
  }
}

/** In-memory Storage double speaking the real get/set protocol. */
class FakeStorage extends Abject {
  readonly data = new Map<string, unknown>();
  constructor() { super({ manifest: iface('Storage') }); }
  protected async onInit(): Promise<void> {
    this.on('get', (msg: AbjectMessage) => {
      const { key } = msg.payload as { key: string };
      return this.data.get(key) ?? null;
    });
    this.on('set', (msg: AbjectMessage) => {
      const { key, value } = msg.payload as { key: string; value: unknown };
      this.data.set(key, value);
      return true;
    });
  }
}

/** Stands in for HttpClient: accepts addDependent (universal) and lets the
 *  test fire a real changed('httpExchange') through the real dependents path. */
class FakeHttpClient extends Abject {
  constructor() { super({ manifest: iface('HttpClient') }); }
  protected async onInit(): Promise<void> {
    this.on('emitExchange', (msg: AbjectMessage) => {
      this.changed('httpExchange', msg.payload);
      return true;
    });
  }
}

/** Test driver: an initialized abject the test uses to send real requests. */
class Driver extends Abject {
  constructor() { super({ manifest: iface('Driver') }); }
  protected async onInit(): Promise<void> {}
  async call<T>(to: AbjectId, method: string, payload: unknown): Promise<T> {
    return this.request<T>(request(this.id, to, method, payload));
  }
}

async function until(cond: () => boolean, ms = 3000): Promise<void> {
  const deadline = Date.now() + ms;
  while (!cond()) {
    if (Date.now() > deadline) throw new Error('condition not met in time');
    await new Promise(r => setTimeout(r, 25));
  }
}

interface Rig {
  registry: FakeRegistry;
  storage: FakeStorage;
  http: FakeHttpClient;
  driver: Driver;
  recorder: CassetteRecorder;
}

async function rig(opts: { registerHttp?: boolean; registerStorage?: boolean } = {}): Promise<Rig> {
  const bus = new MessageBus();
  const registry = new FakeRegistry();
  await registry.init(bus);
  const storage = new FakeStorage();
  const http = new FakeHttpClient();
  const driver = new Driver();
  const recorder = new CassetteRecorder({ flushMs: 10, discoverRetryMs: 20 });
  for (const a of [storage, http, driver, recorder]) await a.init(bus, undefined, registry.id);
  if (opts.registerHttp !== false) registry.addEntry(http.id, 'HttpClient');
  if (opts.registerStorage !== false) registry.addEntry(storage.id, 'Storage');
  if (opts.registerHttp !== false) await recorder.ready();
  return { registry, storage, http, driver, recorder };
}

const CALLER = 'caller-1111-2222-3333-444444444444' as AbjectId;
const CALLER_TYPE = 'peer/ws/user/Fetcher' as TypeId;

function exchange(over: Record<string, unknown> = {}) {
  return {
    caller: CALLER,
    request: { method: 'GET', url: 'https://api.example.com/v1/items?page=2', headers: {} },
    response: { status: 200, headers: {}, bodyText: '{"items":[1,2]}' },
    durationMs: 12,
    at: 1756400000000,
    ...over,
  };
}

test('records an httpExchange as a cassette under the caller typeId', async () => {
  const r = await rig();
  r.registry.addEntry(CALLER, 'Fetcher', CALLER_TYPE);

  await r.driver.call(r.http.id, 'emitExchange', exchange());
  await until(() => r.storage.data.has(`cassettes:${CALLER_TYPE}`));

  const stored = r.storage.data.get(`cassettes:${CALLER_TYPE}`) as Array<Record<string, any>>;
  assert.equal(stored.length, 1);
  const c = stored[0];
  assert.equal(c.method, '_http');
  assert.equal(c.request.method, 'GET');
  assert.equal(c.request.url, 'https://api.example.com/v1/items?page=2');
  assert.equal(c.response.status, 200);
  assert.equal(c.response.bodyText, '{"items":[1,2]}');
  assert.equal(c.at, 1756400000000);
});

test('skips exchanges whose caller has no durable typeId', async () => {
  const r = await rig();
  // CALLER registered with a name but NO typeId
  r.registry.addEntry(CALLER, 'Fetcher');

  await r.driver.call(r.http.id, 'emitExchange', exchange());
  await new Promise(res => setTimeout(res, 150));

  assert.equal(r.storage.data.size, 0);
});

test('merges with cassettes already in Storage instead of clobbering', async () => {
  const r = await rig();
  r.registry.addEntry(CALLER, 'Fetcher', CALLER_TYPE);
  const preexisting = [{
    method: '_http',
    request: { method: 'GET', url: 'https://api.example.com/v1/old', headers: {} },
    response: { status: 200, bodyText: 'old' },
    at: 1756300000000,
  }];
  r.storage.data.set(`cassettes:${CALLER_TYPE}`, preexisting);

  await r.driver.call(r.http.id, 'emitExchange', exchange());
  await until(() => {
    const s = r.storage.data.get(`cassettes:${CALLER_TYPE}`) as unknown[];
    return Array.isArray(s) && s.length === 2;
  });

  const stored = r.storage.data.get(`cassettes:${CALLER_TYPE}`) as Array<Record<string, any>>;
  assert.equal(stored[0].request.url, 'https://api.example.com/v1/old');
  assert.equal(stored[1].request.url, 'https://api.example.com/v1/items?page=2');
});

test('evicts FIFO per endpoint bucket, capped at 5 per bucket', async () => {
  const r = await rig();
  r.registry.addEntry(CALLER, 'Fetcher', CALLER_TYPE);

  for (let i = 0; i < 7; i++) {
    await r.driver.call(r.http.id, 'emitExchange', exchange({
      request: { method: 'GET', url: `https://api.example.com/v1/items?page=${i}`, headers: {} },
      at: 1756400000000 + i,
    }));
  }
  // one exchange on a DIFFERENT endpoint must survive the churn above
  await r.driver.call(r.http.id, 'emitExchange', exchange({
    request: { method: 'GET', url: 'https://api.example.com/v1/users', headers: {} },
    at: 1756400009999,
  }));

  await until(() => {
    const s = r.storage.data.get(`cassettes:${CALLER_TYPE}`) as unknown[];
    return Array.isArray(s) && s.length === 6;
  });

  const stored = r.storage.data.get(`cassettes:${CALLER_TYPE}`) as Array<Record<string, any>>;
  const items = stored.filter(c => (c.request.url as string).includes('/v1/items'));
  const users = stored.filter(c => (c.request.url as string).includes('/v1/users'));
  assert.equal(items.length, 5, 'items bucket capped at 5');
  assert.equal(users.length, 1, 'other bucket untouched');
  // FIFO: the two OLDEST items exchanges are gone
  assert.ok(items.every(c => (c.at as number) >= 1756400000002));
});

test('global cap evicts from the LARGEST bucket, never a rare endpoint\'s only recording', async () => {
  const r = await rig();
  r.registry.addEntry(CALLER, 'Fetcher', CALLER_TYPE);

  // the rare endpoint recorded FIRST (chronologically oldest)
  await r.driver.call(r.http.id, 'emitExchange', exchange({
    request: { method: 'GET', url: 'https://api.example.com/rare', headers: {} }, at: 1756400000000,
  }));
  // one hot bucket at its per-bucket cap of 5
  for (let i = 0; i < 5; i++) {
    await r.driver.call(r.http.id, 'emitExchange', exchange({
      request: { method: 'GET', url: `https://api.example.com/hot?p=${i}`, headers: {} },
      at: 1756400001000 + i,
    }));
  }
  // 45 more singleton endpoints -> total 51, one over the global 50 cap
  for (let i = 0; i < 45; i++) {
    await r.driver.call(r.http.id, 'emitExchange', exchange({
      request: { method: 'GET', url: `https://api.example.com/one-off-${i}`, headers: {} },
      at: 1756400002000 + i,
    }));
  }

  await until(() => {
    const s = r.storage.data.get(`cassettes:${CALLER_TYPE}`) as unknown[];
    return Array.isArray(s) && s.length === 50;
  });
  const stored = r.storage.data.get(`cassettes:${CALLER_TYPE}`) as Array<Record<string, any>>;
  assert.ok(stored.some(c => c.request.url.endsWith('/rare')), 'rare survives despite being oldest');
  assert.equal(stored.filter(c => (c.request.url as string).includes('/hot')).length, 4,
    'overflow came out of the largest bucket');
});

test('subscribes to HttpClient even when it registers late, and keeps retrying', async () => {
  const r = await rig({ registerHttp: false });
  r.registry.addEntry(CALLER, 'Fetcher', CALLER_TYPE);

  await new Promise(res => setTimeout(res, 200)); // recorder is retrying discovery
  r.registry.addEntry(r.http.id, 'HttpClient');
  await r.recorder.ready();

  await r.driver.call(r.http.id, 'emitExchange', exchange());
  await until(() => r.storage.data.has(`cassettes:${CALLER_TYPE}`));
});

test('records while Storage is missing and persists once it appears, merging what the store held', async () => {
  const r = await rig({ registerStorage: false });
  r.registry.addEntry(CALLER, 'Fetcher', CALLER_TYPE);
  // the store already holds evidence from a previous run - must not be clobbered
  r.storage.data.set(`cassettes:${CALLER_TYPE}`, [{
    method: '_http',
    request: { method: 'GET', url: 'https://api.example.com/v1/old', headers: {} },
    response: { status: 200, bodyText: 'old' }, at: 1756300000000,
  }]);

  await r.driver.call(r.http.id, 'emitExchange', exchange());
  await new Promise(res => setTimeout(res, 100)); // recorded in memory, nowhere to persist yet

  r.registry.addEntry(r.storage.id, 'Storage');
  await until(() => {
    const s = r.storage.data.get(`cassettes:${CALLER_TYPE}`) as unknown[];
    return Array.isArray(s) && s.length === 2;
  });
  const stored = r.storage.data.get(`cassettes:${CALLER_TYPE}`) as Array<Record<string, any>>;
  assert.equal(stored[0].request.url, 'https://api.example.com/v1/old');
});

lab/http-emit-unit.ts

/**
 * One-shot verification for HttpClient's httpExchange emission (PR-A).
 * Not committed — goes in the PR description. Run: npx tsx --test lab/http-emit-unit.ts
 *
 * The network edge (makeRequest) is stubbed via subclass because validateDomain
 * blocks loopback (SSRF guard) — everything else is real: real bus, real
 * handlers, real dependents protocol, real redaction.
 */
import { test } from 'node:test';
import assert from 'node:assert/strict';
import { MessageBus } from '../src/runtime/message-bus.js';
import { Abject } from '../src/core/abject.js';
import { request } from '../src/core/message.js';
import type { AbjectId, AbjectMessage } from '../src/core/types.js';
import { HttpClient, HttpRequest, HttpResponse } from '../src/objects/capabilities/http-client.js';

class TestHttpClient extends HttpClient {
  canned: HttpResponse = { status: 200, statusText: 'OK', headers: { 'content-type': 'application/json' }, body: '{"ok":true}', ok: true };
  seen: HttpRequest[] = [];
  override async makeRequest(req: HttpRequest): Promise<HttpResponse> {
    this.seen.push(req);
    return this.canned;
  }
}

const iface = (name: string) => ({
  name,
  description: `test ${name}`,
  version: '0.0.1',
  interface: { id: `test:${name.toLowerCase()}`, name, description: name, methods: [] },
});

class Listener extends Abject {
  readonly events: Array<Record<string, any>> = [];
  constructor() { super({ manifest: iface('Listener') }); }
  protected async onInit(): Promise<void> {
    this.on('httpExchange', (msg: AbjectMessage) => {
      this.events.push(msg.payload as Record<string, any>);
      return true;
    });
  }
  async subscribeTo(target: AbjectId): Promise<void> {
    await this.request(request(this.id, target, 'addDependent', {}));
  }
}

class Driver extends Abject {
  constructor() { super({ manifest: iface('Driver') }); }
  protected async onInit(): Promise<void> {}
  async call<T>(to: AbjectId, method: string, payload: unknown): Promise<T> {
    return this.request<T>(request(this.id, to, method, payload));
  }
}

async function until(cond: () => boolean, ms = 3000): Promise<void> {
  const deadline = Date.now() + ms;
  while (!cond()) {
    if (Date.now() > deadline) throw new Error('condition not met in time');
    await new Promise(r => setTimeout(r, 25));
  }
}

async function rig() {
  const bus = new MessageBus();
  const http = new TestHttpClient();
  const listener = new Listener();
  const driver = new Driver();
  for (const a of [http, listener, driver]) await a.init(bus);
  await listener.subscribeTo(http.id);
  return { http, listener, driver };
}

test('a completed request is emitted to dependents with caller attribution', async () => {
  const r = await rig();
  const reply = await r.driver.call<HttpResponse>(r.http.id, 'request', {
    method: 'GET', url: 'https://api.example.com/v1/items?page=2',
  });
  assert.equal(reply.status, 200);

  await until(() => r.listener.events.length === 1);
  const e = r.listener.events[0];
  assert.equal(e.caller, r.driver.id);
  assert.equal(e.request.method, 'GET');
  assert.equal(e.request.url, 'https://api.example.com/v1/items?page=2');
  assert.equal(e.response.status, 200);
  assert.equal(e.response.bodyText, '{"ok":true}');
  assert.equal(typeof e.at, 'number');
  assert.equal(typeof e.durationMs, 'number');
});

test('secret query params and auth headers are redacted before emission', async () => {
  const r = await rig();
  await r.driver.call(r.http.id, 'request', {
    method: 'GET',
    url: 'https://api.example.com/v1/items?api_key=sk-SECRET&page=2&TOKEN=abc',
    headers: { authorization: 'Bearer sk-SECRET', cookie: 'session=SECRET', accept: 'application/json' },
  });

  await until(() => r.listener.events.length === 1);
  const e = r.listener.events[0];
  assert.ok(!JSON.stringify(e).includes('SECRET'), 'no secret anywhere in the event');
  assert.equal(e.request.url, 'https://api.example.com/v1/items?api_key=REDACTED&page=2&TOKEN=REDACTED');
  assert.equal(e.request.headers.authorization, 'REDACTED');
  assert.equal(e.request.headers.cookie, 'REDACTED');
  assert.equal(e.request.headers.accept, 'application/json');
  // the request handed to the network is NOT redacted
  assert.ok((r.http.seen[0].url).includes('sk-SECRET'));
});

test('response set-cookie is redacted before emission', async () => {
  const r = await rig();
  r.http.canned = { status: 200, statusText: 'OK', headers: { 'set-cookie': 'sid=SECRET', 'content-type': 'text/plain' }, body: 'hi', ok: true };
  await r.driver.call(r.http.id, 'get', { url: 'https://api.example.com/hello' });

  await until(() => r.listener.events.length === 1);
  const e = r.listener.events[0];
  assert.equal(e.response.headers['set-cookie'], 'REDACTED');
  assert.equal(e.response.headers['content-type'], 'text/plain');
});

test('bodies over the cap are truncated with a flag', async () => {
  const r = await rig();
  r.http.canned = { status: 200, statusText: 'OK', headers: {}, body: 'x'.repeat(100_000), ok: true };
  await r.driver.call(r.http.id, 'get', { url: 'https://api.example.com/big' });

  await until(() => r.listener.events.length === 1);
  const e = r.listener.events[0];
  assert.equal(e.response.bodyText.length, 64 * 1024);
  assert.equal(e.response.truncated, true);
});

test('post and postJson entry points emit too, with request body as text', async () => {
  const r = await rig();
  await r.driver.call(r.http.id, 'postJson', {
    url: 'https://api.example.com/v1/items', data: { name: 'widget' },
  });

  await until(() => r.listener.events.length === 1);
  const e = r.listener.events[0];
  assert.equal(e.request.method, 'POST');
  assert.equal(e.request.bodyText, '{"name":"widget"}');
});

test('secret stems are redacted: client_secret, refresh_token, X-Amz-Signature, x-amz-security-token', async () => {
  const r = await rig();
  await r.driver.call(r.http.id, 'request', {
    method: 'GET',
    url: 'https://api.example.com/cb?client_secret=SECRET1&refresh_token=SECRET2&X-Amz-Signature=SECRET3&page=1',
    headers: { 'x-amz-security-token': 'SECRET4', 'x-auth-token': 'SECRET5', accept: 'text/plain' },
  });

  await until(() => r.listener.events.length === 1);
  const e = r.listener.events[0];
  assert.ok(!JSON.stringify(e).match(/SECRET[0-9]/), 'no stem-named secret survives');
  assert.ok(e.request.url.includes('page=1'), 'non-secret params survive');
  assert.equal(e.request.headers.accept, 'text/plain');
});

test('secret-bearing fields inside body text are redacted, JSON and form-encoded', async () => {
  const r = await rig();
  r.http.canned = {
    status: 200, statusText: 'OK', headers: {},
    body: '{"access_token":"sk-BODYSECRET","expires_in":3600,"scope":"read"}', ok: true,
  };
  await r.driver.call(r.http.id, 'request', {
    method: 'POST', url: 'https://api.example.com/oauth/token',
    body: 'grant_type=refresh_token&refresh_token=BODYSECRET2&client_id=pub123',
  });

  await until(() => r.listener.events.length === 1);
  const e = r.listener.events[0];
  assert.ok(!JSON.stringify(e).includes('BODYSECRET'), 'no body secret crosses the bus');
  assert.ok(e.response.bodyText.includes('"expires_in":3600'), 'non-secret JSON fields survive');
  assert.ok(e.request.bodyText.includes('client_id=pub123'), 'non-secret form fields survive');
});

test('nothing is emitted (or serialized) when no dependents are subscribed', async () => {
  const bus2 = new MessageBus();
  const http2 = new TestHttpClient();
  const driver2 = new Driver();
  for (const a of [http2, driver2]) await a.init(bus2);
  let stringified = 0;
  const big = { toJSON() { stringified++; return { x: 1 }; } };
  await driver2.call(http2.id, 'request', { method: 'POST', url: 'https://api.example.com/v1', body: big });
  await new Promise(res => setTimeout(res, 100));
  // makeRequest is stubbed here, so the only serialization that could run is
  // the emission path's - and with no dependents it must not run at all.
  assert.equal(stringified, 0, 'no event serialization when nobody is subscribed');
});

lab/recorder-live.ts

/**
 * End-to-end evidence for the PR-A recorder (mempko/abject#11 series).
 * Not committed — goes in the PR description. Run: npx tsx --test lab/recorder-live.ts
 *
 * Production classes throughout: real Registry, real Storage, real HttpClient
 * (network edge stubbed — validateDomain blocks loopback, so no live socket),
 * real CassetteRecorder, real MessageBus, real dependents protocol. Proves the
 * full lifecycle: boot-style spawn/register -> discovery -> subscription ->
 * a caller's HTTP request -> typeId-keyed cassette readable back from Storage.
 */
import { test } from 'node:test';
import assert from 'node:assert/strict';
import { MessageBus } from '../src/runtime/message-bus.js';
import { Abject } from '../src/core/abject.js';
import { request } from '../src/core/message.js';
import type { AbjectId, TypeId } from '../src/core/types.js';
import { Registry } from '../src/objects/registry.js';
import { Storage } from '../src/objects/capabilities/storage.js';
import { HttpClient, HttpRequest, HttpResponse } from '../src/objects/capabilities/http-client.js';
import { CassetteRecorder } from '../src/objects/cassette-recorder.js';

class StubNetHttpClient extends HttpClient {
  override async makeRequest(_req: HttpRequest): Promise<HttpResponse> {
    return { status: 200, statusText: 'OK', headers: { 'content-type': 'application/json' }, body: '{"price":42}', ok: true };
  }
}

class Caller extends Abject {
  constructor() {
    super({ manifest: { name: 'PriceFetcher', description: 'test caller', version: '0.0.1',
      interface: { id: 'test:pricefetcher', name: 'PriceFetcher', description: 'x', methods: [] },
      requiredCapabilities: [], providedCapabilities: [], tags: [] } });
  }
  protected async onInit(): Promise<void> {}
  async fetchPrice(httpId: AbjectId): Promise<HttpResponse> {
    return this.request<HttpResponse>(request(this.id, httpId, 'get',
      { url: 'https://api.example.com/price?api_key=sk-LIVE-SECRET' }));
  }
  async readStorage<T>(storageId: AbjectId, key: string): Promise<T> {
    return this.request<T>(request(this.id, storageId, 'get', { key }));
  }
}

async function until(cond: () => Promise<boolean>, ms = 4000): Promise<void> {
  const deadline = Date.now() + ms;
  while (!(await cond())) {
    if (Date.now() > deadline) throw new Error('condition not met in time');
    await new Promise(r => setTimeout(r, 50));
  }
}

test('full lifecycle: spawn, subscribe, record, persist, read back', async () => {
  const bus = new MessageBus();
  const registry = new Registry();
  await registry.init(bus);
  const storage = new Storage('lab-recorder-live');
  const http = new StubNetHttpClient();
  const caller = new Caller();
  const recorder = new CassetteRecorder({ flushMs: 20 });
  for (const a of [storage, http, caller, recorder]) await a.init(bus, undefined, registry.id);

  // Boot-style registration, typeIds the way WorkspaceManager stamps them.
  const reg = async (a: Abject, typeId: string) => {
    await (caller as any).request(request(caller.id, registry.id, 'register', {
      objectId: a.id, manifest: (a as any).manifest, typeId: typeId as TypeId,
    }));
  };
  await reg(http, 'peer/system/HttpClient');
  await reg(storage, 'peer/system/Storage');
  await reg(recorder, 'peer/system/CassetteRecorder');
  await reg(caller, 'peer/ws/user/PriceFetcher');

  // Recorder discovers deps and subscribes through the REAL registry.
  await recorder.ready();

  const reply = await caller.fetchPrice(http.id);
  assert.equal(reply.status, 200);

  let stored: any;
  await until(async () => {
    stored = await caller.readStorage(storage.id, 'cassettes:peer/ws/user/PriceFetcher');
    return Array.isArray(stored) && stored.length === 1;
  });

  const c = stored[0];
  assert.equal(c.method, '_http');
  assert.equal(c.request.url, 'https://api.example.com/price?api_key=REDACTED');
  assert.equal(c.response.bodyText, '{"price":42}');
  assert.ok(!JSON.stringify(stored).includes('sk-LIVE-SECRET'));
});

What comes next in the series

The blocklist PR (Atomics/SharedArrayBuffer, its own reasoning), then the judge as an abject, evaluated inside the deploy ops. Once cassettes accumulate here, the learned-schema conversation from #11 has real data to stand on.

[mempko]

Thanks for landing the recorder first, and for keeping the whole path on the bus. The shape is right: changed() as the seam, registry-verified attribution, typeId keys, merge before write. Two things need to change before this merges, one of them a design point.

Only record user-created objects

The purpose of the cassettes is evidence about objects the generator wrote. As built, the recorder captures every direct caller of HttpClient with a typeId, and that includes the system objects. The biggest one is LLMObject: every provider call goes through HttpClient's request handler (createFetchDelegate in llm-object.ts), so with this PR each completion becomes a cassette under cassettes:{peer}/system/LLMObject. That is full prompts, system prompts, chat history, and KnowledgeBase facts, 64K per body, 50 kept, in cleartext in Storage. The name-based redaction removes x-api-key and authorization but nothing inside messages or content matches a stem. WebFetch, WebSearch, WebAgent, OAuthHelper, and the catalog clients are in the same position.

Two problems come with that. First, it breaks the workspace privacy boundary: one pile under a single system key holds private prompts from every workspace. Second, volume: each flush ships the whole list, up to several megabytes, from the recorder's worker to Storage and rewrites one row every 500ms while the LLM is busy, which is the same shape as the per-frame saveData flood we already had to fix once.

Please have the recorder skip system-scoped callers and record only user-created objects. TypeIds already carry the scope: system objects are {peer}/system/{Name}, user objects are {peer}/{workspace}/user/{Name}. Checking the resolved typeId inside record() after resolveCallerIdentity is enough, and it keeps the emission side in HttpClient unchanged. Since system objects never produce cassettes then, a hot LLMObject also stops paying for the redaction and event build on every call, because the recorder is still a dependent. If you would rather the recorder subscribe with a filter so HttpClient skips the work too, that is fine, but keep the policy in the recorder, not in HttpClient.

Boot fails with workers enabled

CassetteRecorder is in the workerEligible list in server/index.ts but its constructor is never registered in workers/abject-worker-node.ts. Workers are on by default (cores minus one, minimum one), so the Factory sends the spawn to a worker, the worker answers No constructor for 'CassetteRecorder' in worker, supervisedSpawn rejects, and main() exits with a fatal startup error. Your scripts construct the recorder directly and never spawn through the Factory, so they cannot see this. Add the import and a constructors.set('CassetteRecorder', ...) line in the worker entry. CLAUDE.md calls this pitfall out under per-workspace objects, and it applies to anything marked worker-eligible.

Smaller items

• Export CassetteRecorder and a CASSETTE_RECORDER_ID constant from src/index.ts, like the other system objects.
• In core/abject.ts the new hasDependents getter landed between changed() and its doc comment, so the note about not registering both notification styles now documents the getter. Move the getter above the comment.
• When Storage is absent, flush() re-arms a 500ms timer per typeId and calls discoverDep('Storage') on every tick. Backing that off the way subscribe() does would be enough.
• The key stem redacts keyword= and idempotency-key. You called out the false-positive tradeoff and I agree with erring toward matching; just noting it so it does not surprise anyone reading a cassette.

Everything else checked out for me: the branch merges cleanly onto current main, tsc --noEmit passes on the merged tree, the Storage and Registry protocols match what the recorder sends, and the eviction and merge logic read correctly. With the user-only filter and the worker registration in place I am happy to merge.

[andreBurnt]

Thanks for the thorough read, Maxim. All five points are addressed in a439f00, pushed to this branch.

Only record user-created objects. record() now checks the resolved typeId after resolveCallerIdentity and returns unless it is the 4-segment {peer}/{workspace}/user/{Name} shape (isUserTypeId() in cassette-recorder.ts). {peer}/system/{Name} and the 3-segment workspace built-ins are skipped, so LLMObject, WebFetch, WebSearch, WebAgent, OAuthHelper and the catalog clients never produce a cassette and never trigger a flush. I kept the policy in the recorder and left HttpClient untouched, as you suggested. Two unit tests cover the system-scoped and built-in cases.

Boot with workers enabled. CassetteRecorder is imported and registered via constructors.set('CassetteRecorder', ...) in workers/abject-worker-node.ts, and I mirrored it in workers/abject-worker.ts so the two constructor tables stay in parity. Verified the worker-eligible spawn path builds it.

Exports. CassetteRecorder, CASSETTE_RECORDER_ID ('abjects:cassette-recorder', same pattern as HEALTH_MONITOR_ID) and the Cassette type are exported from src/index.ts.

hasDependents placement. Moved above the changed() doc comment, so the dual-notification-style note documents changed() again.

Storage discovery backoff. When Storage is absent, flush() now re-arms with a per-recorder delay that starts at flushMs and doubles up to 5000ms, the same shape as subscribe(). The delay resets to flushMs once discoverDep('Storage') succeeds, so the first real flush after Storage appears is not penalized.

Re-verified: tsc --noEmit clean on the branch and on the branch merged onto current main (f06522a, merges clean); recorder unit tests 9/9, HTTP emit tests 8/8, src tests 3/3. Ready for another look.

[mempko]

Thanks for the quick turnaround on a439f00. All four items from the last round are done the way I asked: the user-only filter lives in the recorder and matches the typeId shape AbjectStore stamps, the worker entries both register the constructor, the exports and the getter placement are right, and the Storage backoff is real. I merged the branch onto current main again and tsc --noEmit is clean. Reading the emission side more closely turned up two things that have to change before this merges, and one of them reframes the seam.

Exchanges are broadcast to anyone who asks

addDependent is the universal dependents protocol: any object on the bus can send it, and nothing checks who. As built, HttpClient emits every exchange to every dependent, so a generated object in any workspace can subscribe and receive every other object's HTTP traffic, including LLMObject's prompts and other workspaces' user objects' API calls, with only credential names scrubbed. The user-only filter in the recorder does not touch this, because the filter runs after the broadcast. That is the workspace privacy boundary broken one hop earlier than the Storage pile from the last round.

The fix is to stop treating this as a broadcast. HttpClient should discover the CassetteRecorder through the registry and send each exchange as an event straight to that id, and only that id. System objects knowing about each other is fine; that is what the registry is for. Re-resolve on recipientGone so a recorder restart picks back up, and drop the changed() path for exchanges entirely. That also halves the bus traffic per exchange, since changed() sends both the aspect event and a generic changed copy the recorder ignores, each carrying up to 128K characters across the worker boundary.

Skip LLMObject's calls at the source

Every provider call goes through HttpClient's request handler, and today every one of them pays for redaction and event construction before the recorder drops it. HttpClient should resolve LLMObject from the registry at init and skip emission for that caller. One id comparison per request, no policy about user objects in HttpClient, and LLMObject stops paying anything. If other system callers with large bodies turn up, the same shape covers them.

The redaction regex is superlinear

This still matters with the skip in place, because user objects' bodies go through it. capBody runs redactBodyText over the full body before the 64K cut, and JSON_SECRET_FIELD has the shape "[^"\\]*(?:stem)[^"\\]*"\s*:\s*". When a long string value contains a stem word and is not followed by a colon, the engine backtracks across every way of splitting the string around the stem before it gives up. I ran the two regexes exactly as written in the diff over synthetic JSON bodies whose string values contain "key", "token", and "session":

Body	Time
100 KB	140 ms
300 KB	1.2 s
1 MB	9.7 s
300 KB, no stem words in values	1 ms

That runs synchronously inside the request handler on a shared pool worker, so it stalls every object co-located with HttpClient. Same class of problem as the saveData flood.

The fix is small: match any string key / string value pair with a linear pattern and decide in a replacer, instead of putting the stem alternation inside the key's character class. Same for the form pattern. This keeps the "no JSON.parse on this path" property.

const JSON_STRING_FIELD = /"((?:[^"\\]|\\.)*)"(\s*:\s*")(?:[^"\\]|\\.)*"/g;
const FORM_FIELD = /(^|[&?])([^=&]*=)[^&]*/g;

function redactBodyText(body: string): string {
  return body
    .replace(JSON_STRING_FIELD, (m, key: string, sep: string) =>
      SECRET_NAME_STEM.test(key) ? `"${key}"${sep}REDACTED"` : m)
    .replace(FORM_FIELD, (m, lead: string, name: string) =>
      SECRET_NAME_STEM.test(name) ? `${lead}${name}REDACTED` : m);
}

With that, the same 1 MB body scrubs in 2 ms, and the stem cases from your scripts (api_key, access_token, refresh_token, nested auth.token, author left alone, escaped quotes inside a value) all come out the same. Please re-run http-emit-unit.ts against it and add a body in the hundreds of KB with stem words in the values, so this class of regression has a check.

Noted, not blocking

• If the Supervisor restarts HttpClient, nothing re-establishes the link today. With HttpClient doing the discovery instead of the recorder, a fresh HttpClient resolves the recorder at init and this goes away on its own.
• The recorder keeps every user type's list in memory indefinitely and each flush rewrites the whole list. Fine at current volume.
• Cassettes for workspace objects land in global Storage rather than the workspace's Storage. Keys are scoped so nothing mixes, but the data sits outside the workspace boundary. Worth a conversation before the judge PR reads them.

With the direct-to-recorder seam, the LLMObject skip, and the linear scrub in place, I am ready to merge this.

[andreBurnt]

All three are in 97030bc, pushed to this branch.

Direct-to-recorder seam. HttpClient discovers CassetteRecorder through the registry and sends each exchange as a single httpExchange event to that id, and only that id. The changed() path is gone, so the generic changed copy is gone with it, and the hasDependents getter I had added to Abject is removed too: core/abject.ts no longer differs from main. The recorder no longer subscribes to anything; addDependent on HttpClient gets you nothing. A recipientGone handler drops the cached id so the next exchange re-resolves, which is what picks a restarted recorder back up.

One deviation from "at init": HttpClient is spawned before both LLMObject and CassetteRecorder in server/index.ts, so an init-time discoverDep would return null for both. Resolution is lazy instead, on the first exchange, with a 1s negative-cache backoff so running without a recorder costs at most one registry request per second; concurrent exchanges share one in-flight round-trip, and recipientGone clears the backoff so the restart gap is bounded by the one exchange that bounced.

LLM skipped at the source. Per request it is a Map lookup keyed by msg.routing.from; a caller id seen for the first time is classified once against discoverDep('LLM') on the global registry (manifest name, the same lookup LLMMonitor uses) and the verdict cached for that id. I went with per-id classification rather than one cached LLM id because ids are never reused: if the Supervisor restarts LLMObject it comes back with a fresh id, and a single cached id would silently stop matching with no recipientGone to tell HttpClient, since nothing is ever sent to LLM. It also fails closed: while LLM is not discoverable, nothing is recorded at all rather than recorded on a guess (the recorder would drop an unattributable exchange anyway, so no evidence is lost). Workspace registries are not consulted for this, so a user object naming itself LLM does not excuse itself. No user-object policy in HttpClient; the recorder keeps its typeId filter.

Linear redaction. redactBodyText uses your patterns as written, with the stem test in the replacer. Timings with the same kind of synthetic bodies (stem words throughout the string values):

Body	Before	After
100 KB	845 ms	1 ms
300 KB	7.5 s	1 ms
1 MB	89 s	4 ms

Output is identical on the existing cases (api_key, access_token, refresh_token, nested auth.token, author untouched, escaped quotes inside a value, keyword= / idempotency-key still redacted as discussed). One pre-existing limit worth stating since you are reading the regex closely: only string values are rewritten, so "pin": 1234 or "token": null pass through, same as before; noted in the comment.

http-emit-unit.ts is now 14 tests: the 300 KB stem-word body asserts the scrub result and a sub-500ms wall clock, and a malformed variant (unterminated string, stray quotes) at the same size checks that broken input does not reopen a slow path. New tests also cover that an addDependent subscriber receives nothing, that LLM's request is neither emitted nor serialized, that a stopped-and-respawned LLM with a fresh id is still skipped, that nothing is recorded before LLM is discoverable, and that a stopped-and-respawned recorder is re-discovered after recipientGone.

Re-verified: tsc --noEmit clean on the branch and on the branch merged onto current main (dbac6b1, merges clean); recorder unit 9/9, HTTP emit 14/14, live lifecycle 1/1 on the merged tree. On the "worth a conversation" note about cassettes living in global Storage: agreed that belongs with the judge PR, and I am fine moving them under the workspace's Storage there if that is the direction you want.