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) 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]; +} 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); + } + } +}