From dcb6197b547ade110b04117fa024777093f1fe0c Mon Sep 17 00:00:00 2001 From: Andre Burnt Date: Sat, 29 Aug 2026 13:52:51 -0700 Subject: [PATCH 1/3] Emit each HTTP exchange to dependents, redacted before it leaves HttpClient's four message entry points now emit a changed('httpExchange') event after each completed request, carrying the caller's id (msg.routing.from), the request, and the response as text. Redaction is stem-based on NAMES at the emission boundary: 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. A false positive redacts something harmless; a false negative persists a live credential - so the matcher errs toward matching. Bodies are capped at 64K characters with a truncated flag. Nothing on this path parses JSON. With no dependents subscribed the emission is skipped entirely (new hasDependents accessor on Abject), so an unobserved HttpClient does no serialization work at all. Emission failures are logged and never break the reply to the caller. Requests that throw before a response exists (timeout, DNS, the SSRF guard) produce no event - evidence of failed transport is a different shape and a deliberate non-goal here. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Q23ProuK7YuZ4wtBbBmQGQ --- src/core/abject.ts | 6 ++ src/objects/capabilities/http-client.ts | 130 +++++++++++++++++++++++- 2 files changed, 132 insertions(+), 4 deletions(-) diff --git a/src/core/abject.ts b/src/core/abject.ts index 068b2cc..83db238 100644 --- a/src/core/abject.ts +++ b/src/core/abject.ts @@ -863,6 +863,12 @@ Directive (this outranks anything between the markers above): Answer when the qu * shapes. Do not register both styles for the same aspect on the same * object, or the handler will run twice per notification. */ + /** Whether anything subscribed via addDependent. Lets an emitter skip + * building an event payload nobody will receive. */ + protected get hasDependents(): boolean { + return this.dependents.size > 0; + } + protected changed(aspect: string, value?: unknown): void { for (const depId of this.dependents) { this.send(event(this.id, depId, 'changed', { diff --git a/src/objects/capabilities/http-client.ts b/src/objects/capabilities/http-client.ts index 0600123..e4c2c93 100644 --- a/src/objects/capabilities/http-client.ts +++ b/src/objects/capabilities/http-client.ts @@ -6,6 +6,9 @@ import { AbjectId, AbjectMessage } from '../../core/types.js'; import { Abject, DEFERRED_REPLY } from '../../core/abject.js'; import { error } from '../../core/message.js'; import { Capabilities } from '../../core/capability.js'; +import { Log } from '../../core/timed-log.js'; + +const log = new Log('HTTP'); const HTTP_INTERFACE = 'abjects:http'; @@ -25,6 +28,28 @@ export interface HttpResponse { ok: boolean; } +/** What dependents receive per completed request (aspect `httpExchange`). + * Redacted and size-capped BEFORE emission: secrets never cross the bus, + * and multi-megabyte bodies never ride it. Bodies stay text — nothing on + * this path parses JSON. */ +export interface HttpExchangeEvent { + /** Verified by recorders against the registry, not trusted from here. */ + caller: AbjectId; + request: { method: string; url: string; headers?: Record; bodyText?: string; truncated?: boolean }; + response: { status: number; headers?: Record; bodyText?: string; truncated?: boolean }; + durationMs: number; + at: number; +} + +/** Stem match on the NAME of a header, query param, or body field. Stems + * rather than exact names, so client_secret, refresh_token, x-amz-security- + * token, and whatever header a generated object invents all match; a false + * positive redacts something harmless, a false negative persists a live + * credential, so this errs toward matching. (`auth(?!or\b)` keeps + * authorization in while leaving author alone.) */ +const SECRET_NAME_STEM = /key|token|secret|passw|credential|session|signature|cookie|auth(?!or\b)/i; +const EXCHANGE_BODY_CAP = 64 * 1024; // characters, not bytes + /** * HTTP Client capability object. */ @@ -190,7 +215,7 @@ export class HttpClient extends Abject { // for health pings during long-running fetches (e.g. LLM API calls). this.on('request', async (msg: AbjectMessage) => { const req = msg.payload as HttpRequest; - this.makeRequest(req).then( + this.tracked(msg, req).then( (result) => this.sendDeferredReply(msg, result), (err) => { this.send(error(msg, 'HTTP_ERROR', @@ -206,7 +231,7 @@ export class HttpClient extends Abject { url: string; headers?: Record; }; - this.makeRequest({ method: 'GET', url, headers }).then( + this.tracked(msg, { method: 'GET', url, headers }).then( (result) => this.sendDeferredReply(msg, result), (err) => { this.send(error(msg, 'HTTP_ERROR', @@ -223,7 +248,7 @@ export class HttpClient extends Abject { body: string; headers?: Record; }; - this.makeRequest({ method: 'POST', url, body, headers }).then( + this.tracked(msg, { method: 'POST', url, body, headers }).then( (result) => this.sendDeferredReply(msg, result), (err) => { this.send(error(msg, 'HTTP_ERROR', @@ -255,7 +280,7 @@ export class HttpClient extends Abject { url: string; data: object; }; - this.makeRequest({ + this.tracked(msg, { method: 'POST', url, body: data, @@ -300,6 +325,53 @@ export class HttpClient extends Abject { /** * Make an HTTP request with retry for transient errors. */ + /** + * makeRequest plus the `httpExchange` event dependents subscribe to + * (a CassetteRecorder, typically). Wraps the message entry points only: + * `msg.routing.from` is the caller identity a recorder attributes the + * exchange to. Emission failures never break the request path, and with + * no dependents `changed()` walks an empty set. + */ + private async tracked(msg: AbjectMessage, req: HttpRequest): Promise { + const started = Date.now(); + const result = await this.makeRequest(req); + // Nobody subscribed: skip serializing/redacting a payload nobody hears. + if (!this.hasDependents) return result; + try { + this.emitExchange(msg.routing.from, req, result, Date.now() - started); + } catch (err) { + // recording is best-effort; the reply to the caller is not + log.warn('httpExchange emission failed', err); + } + return result; + } + + private emitExchange(caller: AbjectId, req: HttpRequest, res: HttpResponse, durationMs: number): void { + const reqBody = typeof req.body === 'string' ? req.body + : req.body !== undefined ? JSON.stringify(req.body) : undefined; + const [reqBodyText, reqTruncated] = capBody(reqBody); + const [resBodyText, resTruncated] = capBody(res.body); + const exchange: HttpExchangeEvent = { + caller, + request: { + method: req.method, + url: redactUrl(req.url), + headers: redactHeaders(req.headers), + ...(reqBodyText !== undefined ? { bodyText: reqBodyText } : {}), + ...(reqTruncated ? { truncated: true } : {}), + }, + response: { + status: res.status, + headers: redactHeaders(res.headers), + ...(resBodyText !== undefined ? { bodyText: resBodyText } : {}), + ...(resTruncated ? { truncated: true } : {}), + }, + durationMs, + at: Date.now(), + }; + this.changed('httpExchange', exchange); + } + async makeRequest(req: HttpRequest): Promise { if (this.webDisabled) throw new Error('Web access is disabled. Enable it in Settings > Permissions.'); // Validate URL @@ -585,3 +657,53 @@ Every response has: { status, statusText, headers, body, ok } // Well-known HTTP client ID export const HTTP_CLIENT_ID = 'abjects:http-client' as AbjectId; + +/** Secret-bearing query params and headers are replaced, never dropped: + * the shape of the request stays visible, the credential does not. */ +function redactUrl(rawUrl: string): string { + try { + const u = new URL(rawUrl); + let touched = false; + for (const name of Array.from(u.searchParams.keys())) { + if (SECRET_NAME_STEM.test(name)) { + u.searchParams.set(name, 'REDACTED'); + touched = true; + } + } + return touched ? u.toString() : rawUrl; + } catch { + return rawUrl.split('?')[0]; + } +} + +function redactHeaders(headers?: Record): Record | undefined { + if (!headers) return undefined; + const out: Record = {}; + for (const [name, value] of Object.entries(headers)) { + out[name] = SECRET_NAME_STEM.test(name) ? 'REDACTED' : value; + } + return out; +} + +/** Field-level scrub of body TEXT: the value of any JSON string field or + * form-encoded field whose NAME matches the secret stems is replaced. Plain + * regex over text - no JSON.parse on this path, ever. Pattern-based, so it + * catches named fields (access_token, client_secret, password), not a + * secret embedded in free text under an innocent name. */ +const JSON_SECRET_FIELD = new RegExp( + `("(?:[^"\\\\]*(?:${SECRET_NAME_STEM.source})[^"\\\\]*)"\\s*:\\s*")(?:[^"\\\\]|\\\\.)*(")`, 'gi'); +const FORM_SECRET_FIELD = new RegExp( + `((?:^|[&?])[^=&]*(?:${SECRET_NAME_STEM.source})[^=&]*=)[^&]*`, 'gi'); + +function redactBodyText(body: string): string { + return body + .replace(JSON_SECRET_FIELD, '$1REDACTED$2') + .replace(FORM_SECRET_FIELD, '$1REDACTED'); +} + +function capBody(body: string | undefined): [string | undefined, boolean] { + if (body === undefined) return [undefined, false]; + const scrubbed = redactBodyText(body); + if (scrubbed.length <= EXCHANGE_BODY_CAP) return [scrubbed, false]; + return [scrubbed.slice(0, EXCHANGE_BODY_CAP), true]; +} From c373090b9d783e126a01f8325fe39ca0fd3be8b8 Mon Sep 17 00:00:00 2001 From: Andre Burnt Date: Sat, 29 Aug 2026 13:52:51 -0700 Subject: [PATCH 2/3] Record objects' HTTP traffic as typeId-keyed cassettes A CassetteRecorder abject subscribes to HttpClient's httpExchange events (addDependent, the universal dependents protocol) and persists each exchange under cassettes: in Storage. The caller's identity is resolved against the registry via resolveCallerIdentity - never trusted from the payload - and exchanges whose caller has no durable typeId are not recorded. A live AbjectId dies with its object; the typeId survives restarts, so the evidence does too. Retention is per endpoint bucket (method + path, query stripped, FIFO cap 5) with an overall cap of 50 per typeId. Global overflow always comes out of the LARGEST bucket, so a rare endpoint's only recording survives no matter how old it is. Failure independence: HttpClient discovery retries with capped backoff for as long as the recorder lives - a permanent recorder that silently gives up has failed its one job - and the subscription never waits on Storage. While Storage is missing, recording continues in memory; persistence catches up once it appears, merging what the store already held so a restart never clobbers accumulated evidence. Everything crosses the bus as messages. HttpClient and this recorder may be hosted on different worker threads (both are workerEligible), which is why no in-process seam could work. See mempko/abject#11. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Q23ProuK7YuZ4wtBbBmQGQ --- src/objects/cassette-recorder.ts | 246 +++++++++++++++++++++++++++++++ 1 file changed, 246 insertions(+) create mode 100644 src/objects/cassette-recorder.ts diff --git a/src/objects/cassette-recorder.ts b/src/objects/cassette-recorder.ts new file mode 100644 index 0000000..6ad8e6d --- /dev/null +++ b/src/objects/cassette-recorder.ts @@ -0,0 +1,246 @@ +/** + * CassetteRecorder - accumulates objects' HTTP traffic as replayable evidence. + * + * Subscribes to HttpClient's `httpExchange` events (Smalltalk dependents + * protocol) and persists each exchange as a cassette under the caller's + * durable typeId (`cassettes:` in Storage). A live AbjectId dies with + * its object; the typeId survives restarts, so the evidence does too. + * + * Everything crosses the bus as messages: the subscription (addDependent), + * the exchanges (events), the persistence (Storage requests). HttpClient and + * this recorder may be hosted on different worker threads, so no in-process + * seam would work - see mempko/abject#11. + * + * Caller attribution comes from `resolveCallerIdentity`, which resolves the + * exchange's caller id against the registry rather than trusting anything in + * the payload. Exchanges whose caller has no durable typeId (unregistered or + * anonymous callers) are not recorded: a cassette that cannot be tied to a + * type is evidence about nothing. + * + * Failure independence: HttpClient discovery retries for as long as this + * object lives (a permanent recorder that silently gives up has failed its + * one job), and recording proceeds in memory while Storage is missing, with + * persistence catching up - and merging what the store already held - once + * Storage appears. + * + * This object only records. Judging the evidence is a separate concern (and a + * separate PR in the #11 series). + */ + +import { AbjectId, AbjectMessage, TypeId } from '../core/types.js'; +import { Abject } from '../core/abject.js'; +import { request } from '../core/message.js'; +import { Log } from '../core/timed-log.js'; +import type { HttpExchangeEvent } from './capabilities/http-client.js'; + +const log = new Log('CASSETTE-RECORDER'); + +const CASSETTE_RECORDER_INTERFACE = 'abjects:cassette-recorder'; + +/** What Storage holds under `cassettes:`: one recorded exchange, + * request/response exactly as emitted (see HttpExchangeEvent - already + * redacted and capped at the emission boundary). `method: '_http'` marks + * raw transport-level evidence, as opposed to a manifest-method invocation. */ +export interface Cassette { + method: '_http'; + request: HttpExchangeEvent['request']; + response: HttpExchangeEvent['response']; + durationMs?: number; + at: number; +} + +/** Per-endpoint retention: a hot parameterized endpoint must not evict the + * one recording of a rarely-hit endpoint, so the per-bucket cap trims hot + * buckets, and global overflow always comes out of the LARGEST bucket. */ +const PER_BUCKET_CAP = 5; +const PER_TYPE_CAP = 50; + +export class CassetteRecorder extends Abject { + private httpClientId?: AbjectId; + private storageId?: AbjectId; + private byType = new Map(); + /** typeIds whose in-memory list has been merged with what Storage held. */ + private merged = new Set(); + private flushTimers = new Map>(); + /** Single FIFO pump: exchanges append in arrival order regardless of how + * long identity resolution or the initial Storage load takes. */ + private pump: Promise = Promise.resolve(); + private readonly flushMs: number; + private readonly discoverRetryMs: number; + private _ready?: Promise; + + constructor(options: { flushMs?: number; discoverRetryMs?: number } = {}) { + super({ + manifest: { + name: 'CassetteRecorder', + description: + 'Records objects\' HTTP traffic as cassettes in Storage, keyed by the caller\'s durable typeId. Subscribes to HttpClient httpExchange events; evidence for judging generated objects accumulates here.', + version: '1.0.0', + interface: { + id: CASSETTE_RECORDER_INTERFACE, + name: 'CassetteRecorder', + description: 'HTTP traffic recording', + methods: [], + }, + requiredCapabilities: [], + providedCapabilities: [], + tags: ['system', 'recording', 'evidence'], + }, + }); + this.flushMs = options.flushMs ?? 500; + this.discoverRetryMs = options.discoverRetryMs ?? 250; + } + + protected override async onInit(): Promise { + this.on('httpExchange', (msg: AbjectMessage) => { + const exchange = msg.payload as HttpExchangeEvent; + this.pump = this.pump + .then(() => this.record(exchange)) + .catch((err) => log.error('record failed', err)); + return true; + }); + this._ready = this.subscribe(); + } + + /** Resolved once the recorder is subscribed to HttpClient. */ + async ready(): Promise { + await this._ready; + } + + /** Subscribe to HttpClient the moment it is discoverable, waiting on + * nothing else - Storage being slow must not delay the subscription, and + * a slow boot must not turn into a permanent silent no-record. Retries + * with capped backoff for as long as this object lives. */ + private async subscribe(): Promise { + let delay = this.discoverRetryMs; + while (!this.httpClientId && (this._status as string) !== 'stopped') { + this.httpClientId = (await this.discoverDep('HttpClient')) ?? undefined; + if (this.httpClientId) break; + await new Promise(r => setTimeout(r, delay)); + delay = Math.min(delay * 2, 5000); + } + if (!this.httpClientId) return; // stopped before HttpClient ever appeared + await this.request(request(this.id, this.httpClientId, 'addDependent', {})); + log.info('subscribed to HttpClient exchanges'); + } + + private async record(exchange: HttpExchangeEvent): Promise { + const identity = await this.resolveCallerIdentity(exchange.caller); + const typeId = identity?.typeId; + if (!typeId) return; + + await this.mergeFromStore(typeId); + const list = this.byType.get(typeId) ?? []; + this.byType.set(typeId, list); + list.push({ + method: '_http', + request: exchange.request, + response: exchange.response, + durationMs: exchange.durationMs, + at: exchange.at, + }); + this.evict(list); + this.scheduleFlush(typeId); + } + + /** The first contact with Storage for a typeId merges what the store + * already holds ahead of this session's recordings - a recorder restart + * must never clobber accumulated evidence. While Storage is missing this + * stays pending and recording continues in memory; the merge happens at + * whichever comes first, the next record or the flush that finds Storage. */ + private async mergeFromStore(typeId: TypeId): Promise { + if (this.merged.has(typeId)) return; + this.storageId ??= (await this.discoverDep('Storage')) ?? undefined; + if (!this.storageId) return; + let existing: Cassette[] = []; + try { + const raw = await this.request( + request(this.id, this.storageId, 'get', { key: `cassettes:${typeId}` }), 5000); + existing = Array.isArray(raw) ? (raw as Cassette[]) : []; + } catch (err) { + log.warn(`existing cassettes unreadable for ${typeId}`, err); + return; // try again next time rather than risk clobbering + } + const current = this.byType.get(typeId) ?? []; + const list = [...existing, ...current]; + this.evict(list); + this.byType.set(typeId, list); + this.merged.add(typeId); + } + + /** Endpoint bucket: method + URL with the query stripped, so parameterized + * hits on one route churn their own bucket only. */ + private bucketKey(c: Cassette): string { + let path = c.request.url; + try { + const u = new URL(c.request.url); + path = u.origin + u.pathname; + } catch { + path = c.request.url.split('?')[0]; + } + return `${c.request.method} ${path}`; + } + + /** FIFO within each over-cap bucket first; then global overflow comes out + * of the LARGEST bucket's oldest entry, so a rare endpoint's only + * recording survives no matter how old it is. In place: the array + * instance is the store. */ + private evict(list: Cassette[]): void { + const counts = new Map(); + for (const c of list) { + const key = this.bucketKey(c); + counts.set(key, (counts.get(key) ?? 0) + 1); + } + for (let i = 0; i < list.length;) { + const key = this.bucketKey(list[i]); + const n = counts.get(key)!; + if (n > PER_BUCKET_CAP) { + counts.set(key, n - 1); + list.splice(i, 1); + } else { + i++; + } + } + while (list.length > PER_TYPE_CAP) { + let maxKey = ''; + let maxN = 0; + for (const [key, n] of counts) { + if (n > maxN) { maxN = n; maxKey = key; } + } + const victim = list.findIndex(c => this.bucketKey(c) === maxKey); + if (victim < 0) break; // counts out of sync - never loop forever + counts.set(maxKey, maxN - 1); + list.splice(victim, 1); + } + } + + private scheduleFlush(typeId: TypeId): void { + const existing = this.flushTimers.get(typeId); + if (existing) clearTimeout(existing); + this.flushTimers.set(typeId, setTimeout(() => { + this.flushTimers.delete(typeId); + void this.flush(typeId); + }, this.flushMs)); + } + + private async flush(typeId: TypeId): Promise { + // Storage may have appeared since recording started; the merge guard + // inside mergeFromStore keeps this idempotent. + await this.mergeFromStore(typeId); + const list = this.byType.get(typeId); + if (!list) return; + if (!this.storageId || !this.merged.has(typeId)) { + // Nowhere to persist yet (or nothing merged yet): keep the data in + // memory and try again - never write a list that might clobber. + this.scheduleFlush(typeId); + return; + } + try { + await this.request( + request(this.id, this.storageId, 'set', { key: `cassettes:${typeId}`, value: list }), 5000); + } catch (err) { + log.warn(`cassette flush failed for ${typeId}`, err); + this.scheduleFlush(typeId); + } + } +} From 7d76b88442cb4f05584a00d4804277a44eebc07d Mon Sep 17 00:00:00 2001 From: Andre Burnt Date: Sat, 29 Aug 2026 13:52:51 -0700 Subject: [PATCH 3/3] Spawn the CassetteRecorder at boot, permanent and worker-eligible Same shape as HealthMonitor: registered constructor, workerEligible, supervisedSpawn permanent with a system typeId. Recording is on from boot - evidence accumulates without anyone asking for it. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Q23ProuK7YuZ4wtBbBmQGQ --- server/index.ts | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/server/index.ts b/server/index.ts index e845773..079d01d 100644 --- a/server/index.ts +++ b/server/index.ts @@ -14,6 +14,7 @@ import { ObjectCreator } from '../src/objects/object-creator.js'; import { ProxyGenerator } from '../src/objects/proxy-generator.js'; import { Negotiator } from '../src/protocol/negotiator.js'; import { HealthMonitor } from '../src/protocol/health-monitor.js'; +import { CassetteRecorder } from '../src/objects/cassette-recorder.js'; import { HttpClient } from '../src/objects/capabilities/http-client.js'; import { NodeStorage } from './node-storage.js'; import { Timer } from '../src/objects/capabilities/timer.js'; @@ -544,6 +545,7 @@ async function main(): Promise { runtime.objectFactory.registerConstructor('ProxyGenerator', () => new ProxyGenerator()); runtime.objectFactory.registerConstructor('Negotiator', () => new Negotiator()); runtime.objectFactory.registerConstructor('HealthMonitor', () => new HealthMonitor()); + runtime.objectFactory.registerConstructor('CassetteRecorder', () => new CassetteRecorder()); runtime.objectFactory.registerConstructor('ObjectCreator', () => new ObjectCreator()); runtime.objectFactory.registerConstructor('AbjectEditor', () => new AbjectEditor()); runtime.objectFactory.registerConstructor('Settings', () => new Settings()); @@ -646,7 +648,7 @@ async function main(): Promise { // Global services 'GlobalSettings', 'PermissionBroker', 'PeerNetwork', 'ObjectCatalog', 'ObjectBrowser', 'MethodInspector', 'ProcessExplorer', 'LLMMonitor', - 'ProxyGenerator', 'Negotiator', 'HealthMonitor', + 'ProxyGenerator', 'Negotiator', 'HealthMonitor', 'CassetteRecorder', 'SkillRegistry', 'SkillBrowser', 'MCPRegistryClient', 'ClawHubClient', 'CatalogBrowser', 'SecretsVault', 'OAuthHelper', @@ -998,6 +1000,10 @@ async function main(): Promise { const proxyGenId = await supervisedSpawn('ProxyGenerator', 'permanent', systemTypeId('ProxyGenerator')); const negotiatorId = await supervisedSpawn('Negotiator', 'permanent', systemTypeId('Negotiator')); const healthMonitorId = await supervisedSpawn('HealthMonitor', 'permanent', systemTypeId('HealthMonitor')); + // Records objects' HTTP traffic as typeId-keyed cassettes (evidence for + // judging generated objects — mempko/abject#11 series). Subscribes to + // HttpClient's httpExchange events at init; recording is on from boot. + const cassetteRecorderId = await supervisedSpawn('CassetteRecorder', 'permanent', systemTypeId('CassetteRecorder')); // Sidebar owns the dock window the rails populate; WorkspaceSwitcher is a // global UI (never hidden during workspace switch)