diff --git a/docs-site/static/search-index.json b/docs-site/static/search-index.json index d55fd9623..20cf4d648 100644 --- a/docs-site/static/search-index.json +++ b/docs-site/static/search-index.json @@ -8368,7 +8368,7 @@ {"objectID":"549e7b7594cf096b8e419d967e9cfa997bb10defbc439635f2152bc7a919b3e5","title":"5. Behavioral contract (acceptance spec)","url":"/docs/plans/2026-07-27-isolated-agent-runner-rfc#5-behavioral-contract-acceptance-spec","content":"Lifecycle — worker instance from N3, disposed in — including on the leg\n path when the handle expires (TTL) or is stopped.\nResearch pass runs with tools under the turn budget ( + wrap-up nudge\nstall watchdog — the machinery from , not a reimplementation). NEVER a\n bare wall-clock abort: a budget-capped run ends with the model consolidating, an honest\n , and . (Leashed is fed in as \n so leg ends are consolidations too; only and waste trips end a leg by\n abort, and both preserve the records.)\nExtraction pass always runs (when is configured), tools disabled, on\n its OWN timeout — never carved out of the research budget — fed from the tool-execution\n records, so a research generate that died on a provider error still extracts from the\n records instead of losing the run. In leashed mode extraction runs on terminal legs;\n intermediate legs return summaries instead (per the O2 pattern —\n extracting every leg would burn the budget the leash exists to protect).\nStructured recovery built in — candidates in order: provider → raw\n JSON object → coerceschemamaxRetriesshapeDocstatus: 'error' | 'partial'abortSignalstopReason: \"aborted\"onEventstarttoolcalltoolresultphasewrapupleg_endwastecompleteerrortoolExecutionCapture.onRecordnextPlanwasteSignalsduplicateCallLimit: 2emptyResultStreakLimit: 3errorStreakLimit: 3noNewResultsLimit: 8`).","hierarchy":{"lvl0":"Plans","lvl1":"RFC: `runIsolatedAgent()` — production sub-agent runner + host-loop delegation","lvl2":"5. Behavioral contract (acceptance spec)","lvl3":""}}, {"objectID":"47dc3f7e12e98eb3908b16a7756ff38b9981a247958227b5a8737bd23da37fbf","title":"6. Field validation matrix","url":"/docs/plans/2026-07-27-isolated-agent-runner-rfc#6-field-validation-matrix","content":"Every public field mapped against the three reference implementations; fields that don't\nmap to all three are justified below the table.\n\n| Field | (a) research worker | (b) log-analysis worker | (c) Yama ContextExplorer |\n| ------------------------------------------------------- | -------------------------------------------------- | ----------------------------------------------------------------------------------------------- | ------------------------ |\n| | persona + code-search MCP filter | persona + log tools | persona + repo tools |\n| (lenient) | evidence list validator | findings validator | context-pack validator |\n| (strict) | provider-attached; no defaults/catch | same | same |\n| | corrective re-ask shape doc | same | same |\n| | wraps bare top-level arrays | normalizes evidence rows | wraps arrays |\n| | parent turn cancels worker | same | same |\n| | O2 defaults | internal-caller overrides (the named co","hierarchy":{"lvl0":"Plans","lvl1":"RFC: `runIsolatedAgent()` — production sub-agent runner + host-loop delegation","lvl2":"6. Field validation matrix","lvl3":""}}, {"objectID":"97996d2242f9f46d729177904ae2aad1103120154667acc5dc5a9f3990601621","title":"6b. Migration note — GenerateResult.toolExecutions shape change","url":"/docs/plans/2026-07-27-isolated-agent-runner-rfc#6b-migration-note-generateresulttoolexecutions-shape-change","content":"The historical stub entries were (with fabricated\n). They are replaced by real s. Consumers\nreading the old fields get — update reads as follows:\n\n| Old field | New field | Notes |\n| ---------- | ------------ | ------------------------------------------------- |\n| | | |\n| | | as parsed by the loop |\n| | | serialized + bounded (~8KB default; cap raisable) |\n| | | real wall-clock, no longer always 0 |\n| — | | new: thrown errors AND error-shaped results |\n| — | | new: epoch ms |\n\n is deliberately unchanged (legacy\n summaries) — the stream-side contract is a\nseparate surface and migrating it is out of scope here.","hierarchy":{"lvl0":"Plans","lvl1":"RFC: `runIsolatedAgent()` — production sub-agent runner + host-loop delegation","lvl2":"6b. Migration note — GenerateResult.toolExecutions shape change","lvl3":""}}, -{"objectID":"e6a7791de93faea3f0d6b1bcad40aa5de97cfdfefacd9b9e8a7d2965f946b67e","title":"6c. Known limitations (follow-up work)","url":"/docs/plans/2026-07-27-isolated-agent-runner-rfc#6c-known-limitations-follow-up-work","content":"Log-bridge attribution: the NeuroLink logger is process-global with a\n single active emitter, so a worker's bridge receives all NeuroLink\n log events in the process, stamped with the bridge's tag. Per-instance\n attribution requires per-instance logger routing.\n/ events fire post-execution (driven by the\n capture record) — a pre-execution hook on the recorder wrapper is the\n natural extension when live in-flight status is needed.\nNested delegations bypass the concurrency pool (the outer delegation\n already holds a slot; queueing nested work behind a full pool would\n deadlock it). Nested fan-out is therefore bounded by the outer slots ×\n per-run step caps, not by the pool directly. standalone\n delegations are always top-level and stay pooled.","hierarchy":{"lvl0":"Plans","lvl1":"RFC: `runIsolatedAgent()` — production sub-agent runner + host-loop delegation","lvl2":"6c. Known limitations (follow-up work)","lvl3":""}}, +{"objectID":"e6a7791de93faea3f0d6b1bcad40aa5de97cfdfefacd9b9e8a7d2965f946b67e","title":"6c. Known limitations (follow-up work)","url":"/docs/plans/2026-07-27-isolated-agent-runner-rfc#6c-known-limitations-follow-up-work","content":"~~Log-bridge attribution~~ — resolved. The logger now routes per\n instance: // run their bodies inside an\n AsyncLocalStorage scope carrying the emitting instance's id, and a worker's\n bridge is subscribed to that id instead of the host's process-wide\n emitter. Residual scope gaps, documented on :\n logs emitted while a consumer drains a returned stream run in the\n consumer's context, and logs emitted outside any call (construction,\n background MCP reconnects) stay unattributed.\n/ events fire post-execution (driven by the\n capture record) — a pre-execution hook on the recorder wrapper is the\n natural extension when live in-flight status is needed.\nNested delegations bypass the concurrency pool (the outer delegation\n already holds a slot; queueing nested work behind a full pool would\n deadlock it). Nested fan-out is therefore bounded by the outer slots ×\n per-run step caps, not by the pool directly. standalone\n delegations are always top-level and stay pooled.","hierarchy":{"lvl0":"Plans","lvl1":"RFC: `runIsolatedAgent()` — production sub-agent runner + host-loop delegation","lvl2":"6c. Known limitations (follow-up work)","lvl3":""}}, {"objectID":"98a6085557f7953269c6b8e5c9e872254ce137f4b6d4933dcca1db421e1fc7d8","title":"7. Implementation notes","url":"/docs/plans/2026-07-27-isolated-agent-runner-rfc#7-implementation-notes","content":"N2 capture rides a per-call \n () attached to the request options and wrapped at\n the two central tool-assembly points (, ) —\n BEFORE tool discovery, so mid-turn hydration inserts wrapped tools. All\n loops (AI-SDK, native Vertex Gemini/Claude, AI Studio) obtain tools through those\n points, so one wrapper covers every path with no per-loop double-recording.\nN3 sampling strip is registry-driven ( /\n in ): an explicit\n on a registry entry wins; otherwise known rejecting-family\n patterns (Sonnet 5, Opus 4.7+, Opus 5, Fable/Mythos) decide. Applied at every\n request-build site whose object retry paths spread ( on both Vertex\n Claude loops, on the OpenAI-compatible path, both direct-Anthropic builders),\n so retries/fallbacks inherit the strip; the reactive\n retry remains as the safety net. A debug log is emitted\n whenever params are stripped.\nN5 turn counting uses AsyncLocalStorage entered at the public when\n agent tools are registered; nested/internal generates share the top-level turn's\n counters ( returns null inside an active scope).\nThe stretch item (neutralizing instruction-shaped patterns in agent reports) is NOT part\n of this change, per the work plan (separate optional PR).","hierarchy":{"lvl0":"Plans","lvl1":"RFC: `runIsolatedAgent()` — production sub-agent runner + host-loop delegation","lvl2":"7. Implementation notes","lvl3":""}}, {"objectID":"bdfafc0f3a8fd37a6271bd7957aff7145529894007854c3fe8ddf4556f2f759f","title":"8. PR slicing","url":"/docs/plans/2026-07-27-isolated-agent-runner-rfc#8-pr-slicing","content":"The work lands as one PR per work item: N1 (agent plumbing), N2 (toolExecutions records),\nN3 (worker factory + sampling strip), N4 (runIsolatedAgent + this RFC), N5 (host-loop\ndelegation + AgentNetwork composition). N4/N5 ship with this RFC in-tree.","hierarchy":{"lvl0":"Plans","lvl1":"RFC: `runIsolatedAgent()` — production sub-agent runner + host-loop delegation","lvl2":"8. PR slicing","lvl3":""}}, {"objectID":"f0fee3d8005091330b88abba8bf41bc4b4e1b5f294425e4f0a250f93bb8f165d","title":"Completing the ai-sdk removal","url":"/docs/plans/2026-09-03-completing-the-ai-sdk-removal","content":"Completing the ai-sdk removal\n\nStatus: done. All 8 items landed. Item 7 — deleting —\nwas the last, in . and are gone from\nand , and keeps\nthem out.\n\nWhat is left, how each piece is solved, and the order forced by their\ndependencies. Every claim here was checked against the source or the installed\npackage, not inferred.\n\nThe rule that governs all of it\n\nRun on every step, not just the live matrix. The live\nmatrix has 40 cells across six providers and it passed a change that broke ten\nof them, because every provider reachable from this machine supports streaming\nand the mocked gate is the only thing that serves a non-streaming body. The\nlive matrix proves behaviour; the mocked gate proves the wire.\nAnthropic native generate\n\n hardcodes when it calls\n (), which is why routing generate through\nit changed the wire.\n\nSolution is the shape already proven for the OpenAI-compatible family: loop over\nthe provider's own delegating-model , which issues a non-streaming\n. Wrap each step in , funnel failures\nthrough , run the turn inside\n (now protected), and call .\n\nCarry over the two fixes found the first time. A schema arriving with no tools\nmust declare as the turn's only tool and pin to it,\nbecause declines an empty tool list. And must\nbe fired explicitly.\nSageMaker\n\n makes one call and already\nreturns ; no streaming is involved, so the wire hazard does not\napply. Same loop shape as above.\n\nThis machine has no SageMaker endpoint or credentials. Its single-step\nbehaviour is identical by construction because it is the same \ncall; the multi-step branch is the new code and needs a live endpoint before it\nis trusted. Say so in the commit rather than implying coverage.\nGuardrails filter and video-analysis formatting\n\nBoth want a single no-tool turn. did this cleanly before\nthe revert took it with everything else: it calls directly and\nreads the text out of the v3 content array. Re-apply unchanged.\nwrapLanguageModel\n\nUpstream is about fifty lines: reverse the middleware array, reduce it, and\nreturn an object that keeps , , and\n while routing and through\n plus the optional / hooks. One\nconsumer, . Reimplement directly.\n\nRecord while doing it that never runs today, because every\nstreaming path is native and bypasses the wrapped model.\nThe tool and schema type algebra\n\nThis is the one that must move as a unit, and the reason the first attempt\nfailed: was replaced while still came from , so the\nreplacement had to satisfy a type it no longer matched.\n\nThe algebra is small and fully specified in the installed package:\nis .\nis the union of , , and\n .\nis a conditional chain over those four.\ncarries , which is\n what ties 's first parameter to the schema.\n\nDeclare all of it in , repoint at the local\ndeclarations, and only then implement , and in\n. is identity upstream and is\n; the work is entirely in the types.\n\n should keep stamping . Nothing in\nthis repo reads it — looks for the plain \nproperty — but keeping it costs nothing and preserves recognition by anything\nthat does.\nThe rest of the public type surface\n\nThree files hand-declare structurally: (the message and part\ntypes), (the model, usage and finish-reason types) and\n (the middleware contract plus the protocol\ntypes from ).\n\nThe non-obvious one is , which embeds inferred\n references. No source edit removes those; they\ndisappear only once itself is local. Verify by hiding the package and\nre-running a consumer typecheck, which is how the leak was confirmed in the\nfirst place.\nGenerationHandler\n\n1409 lines, and its is unreachable once Anthropic and\nSageMaker are native. Four result-formatting helpers on it are still called from\n, but they take the result shape, so they die with\nit. Confirm by trapping the seam and running the full matrix plus the mocked\ngate, then delete. and lose their last consumers here.\nBrowser bundle\n\n still re-exports , ,\n and through the seam. No consumer was found for\nthem. Either drop them or map them onto NeuroLink's own equivalents; this is a\npublic-subpath decision, not a technical one.\n\nOrder\n\nType algebra last among the type work but before the package drop; everything\nelse is independent.\nAnthropic, SageMaker, guardrails and video-analysis — the last runtime\n callers of .\n.\nDelete the dead GenerationHandler path.\nThe tool/schema algebra together with and .\nThe remaining public types.\nBrowser re-exports.\nDrop and , extend , and rewrite the\n documentation snippets that still tell users to import from directly.","hierarchy":{"lvl0":"Plans","lvl1":"Completing the ai-sdk removal","lvl2":"","lvl3":""}}, diff --git a/docs/api/README.md b/docs/api/README.md index a828dc20a..4457ef43f 100644 --- a/docs/api/README.md +++ b/docs/api/README.md @@ -3077,6 +3077,7 @@ console.log(result.content); - [APIValidationResult](type-aliases/APIValidationResult.md) - [ParsedProxyConfig](type-aliases/ParsedProxyConfig.md) - [LogEntry](type-aliases/LogEntry.md) +- [LogEventEmitter](type-aliases/LogEventEmitter.md) - [Logger](type-aliases/Logger.md) - [StructuredError](type-aliases/StructuredError.md) - [EnhancementType](type-aliases/EnhancementType.md) diff --git a/docs/api/type-aliases/LogEventEmitter.md b/docs/api/type-aliases/LogEventEmitter.md new file mode 100644 index 000000000..0b79aadd4 --- /dev/null +++ b/docs/api/type-aliases/LogEventEmitter.md @@ -0,0 +1,34 @@ +[**NeuroLink API Reference**](../README.md) + +--- + +[NeuroLink API Reference](../README.md) / LogEventEmitter + +# Type Alias: LogEventEmitter + +> **LogEventEmitter** = `object` + +Minimal emitter shape the logger forwards `"log-event"` payloads to. + +Structurally satisfied by `node:events`' `EventEmitter` and by NeuroLink's +own typed emitter, so a sink can be either. + +## Properties + +### emit + +> **emit**: (`event`, ...`args`) => `boolean` + +#### Parameters + +##### event + +`string` + +##### args + +...`unknown`[] + +#### Returns + +`boolean` diff --git a/docs/api/type-aliases/Logger.md b/docs/api/type-aliases/Logger.md index d1b3e0621..934006903 100644 --- a/docs/api/type-aliases/Logger.md +++ b/docs/api/type-aliases/Logger.md @@ -11,6 +11,12 @@ Logger interface matching the logger object shape Used for SDK tool contexts and other components that need a logger +Deliberately a subset of the real logger: the per-instance routing methods +(`runInInstanceScope`, `addScopedEventEmitter`, …) are internal plumbing +between the SDK entry points and the logger, and this type is a structural +contract a caller can satisfy — `SDKToolContext.logger`. Adding required +members here would break anyone constructing that context themselves. + ## Properties ### debug @@ -159,9 +165,7 @@ Used for SDK tool contexts and other components that need a logger ##### emitter -###### emit - -(`event`, ...`args`) => `boolean` +[`LogEventEmitter`](LogEventEmitter.md) #### Returns diff --git a/docs/api/type-aliases/WorkerInstanceOptions.md b/docs/api/type-aliases/WorkerInstanceOptions.md index 17bceb8f8..e84f2ef68 100644 --- a/docs/api/type-aliases/WorkerInstanceOptions.md +++ b/docs/api/type-aliases/WorkerInstanceOptions.md @@ -33,13 +33,23 @@ Tag stamped on every forwarded log event (default "worker"). Log bridge sink. Fire-and-forget: listener errors never disrupt the worker. -KNOWN LIMITATION: the underlying NeuroLink logger is process-global with -a single active emitter, so this bridge receives ALL NeuroLink log -events emitted in the process (host, other workers, MCP) for the -worker's lifetime, each stamped with this worker's `tag`. The tag -identifies which bridge forwarded the event, not which instance emitted -it. Per-instance log attribution requires per-instance logger routing — -tracked as follow-up work in the RFC. +Receives only log events this worker emitted. The logger routes per +instance: the worker's `generate` / `stream` / `generateText` run their +bodies inside an AsyncLocalStorage scope carrying the worker's id, and +this bridge is subscribed to that id — so a sibling worker's, the host's +or a background MCP reconnect's logs never arrive here, and `tag` is +true attribution rather than "whichever bridge forwarded it". + +Two things stay outside the scope, by construction: + +- Logs emitted while a **consumer drains a returned stream**. Iteration + happens in the consumer's async context, after `stream()` resolved. +- Logs emitted **outside any call** — construction, background MCP + reconnects, module init. These are unattributed rather than charged to + an arbitrary instance. + +A process-wide sink (`logger.setEventEmitter`) still sees everything, so +a host bridge is unaffected by this narrowing. #### Parameters diff --git a/docs/api/variables/logger.md b/docs/api/variables/logger.md index 5eb0eacfd..8d1451781 100644 --- a/docs/api/variables/logger.md +++ b/docs/api/variables/logger.md @@ -191,9 +191,7 @@ Features: ##### emitter -###### emit - -(`event`, ...`args`) => `boolean` +[`LogEventEmitter`](../type-aliases/LogEventEmitter.md) #### Returns @@ -207,9 +205,89 @@ Features: ##### ifEmitter? -###### emit +[`LogEventEmitter`](../type-aliases/LogEventEmitter.md) + +#### Returns + +`void` + +### runInInstanceScope + +> **runInInstanceScope**: \<`T`\>(`instanceId`, `fn`) => `T` + +#### Type Parameters + +##### T + +`T` + +#### Parameters + +##### instanceId + +`string` + +##### fn + +() => `T` + +#### Returns + +`T` + +### getInstanceScope + +> **getInstanceScope**: () => `string` \| `undefined` + +#### Returns + +`string` \| `undefined` + +### addScopedEventEmitter + +> **addScopedEventEmitter**: (`instanceId`, `emitter`) => `void` + +#### Parameters + +##### instanceId + +`string` + +##### emitter + +[`LogEventEmitter`](../type-aliases/LogEventEmitter.md) + +#### Returns + +`void` + +### removeScopedEventEmitter + +> **removeScopedEventEmitter**: (`instanceId`, `emitter`) => `void` + +#### Parameters + +##### instanceId + +`string` + +##### emitter + +[`LogEventEmitter`](../type-aliases/LogEventEmitter.md) + +#### Returns + +`void` + +### clearScopedEventEmitters + +> **clearScopedEventEmitters**: (`instanceId`) => `void` + +#### Parameters + +##### instanceId -(`event`, ...`args`) => `boolean` +`string` #### Returns diff --git a/docs/plans/2026-07-27-isolated-agent-runner-rfc.md b/docs/plans/2026-07-27-isolated-agent-runner-rfc.md index dfdf2b3b2..646e58328 100644 --- a/docs/plans/2026-07-27-isolated-agent-runner-rfc.md +++ b/docs/plans/2026-07-27-isolated-agent-runner-rfc.md @@ -218,10 +218,14 @@ separate surface and migrating it is out of scope here. ## 6c. Known limitations (follow-up work) -- **Log-bridge attribution**: the NeuroLink logger is process-global with a - single active emitter, so a worker's `onLog` bridge receives all NeuroLink - log events in the process, stamped with the bridge's tag. Per-instance - attribution requires per-instance logger routing. +- ~~**Log-bridge attribution**~~ — **resolved.** The logger now routes per + instance: `generate`/`stream`/`generateText` run their bodies inside an + AsyncLocalStorage scope carrying the emitting instance's id, and a worker's + `onLog` bridge is subscribed to that id instead of the host's process-wide + emitter. Residual scope gaps, documented on `WorkerInstanceOptions.onLog`: + logs emitted while a consumer drains a returned stream run in the + consumer's context, and logs emitted outside any call (construction, + background MCP reconnects) stay unattributed. - **`tool_call`/`tool_result` events fire post-execution** (driven by the capture record) — a pre-execution hook on the recorder wrapper is the natural extension when live in-flight status is needed. diff --git a/package.json b/package.json index 39657861a..0768e77bc 100644 --- a/package.json +++ b/package.json @@ -95,6 +95,7 @@ "test:media": "pnpm exec tsx test/continuous-test-suite-media-gen.ts", "test:media-registry-collisions": "pnpm exec tsx test/continuous-test-suite-media-registry-collisions.ts", "test:memory": "pnpm exec tsx test/continuous-test-suite-memory.ts", + "test:logger-routing": "pnpm exec tsx test/continuous-test-suite-logger-instance-routing.ts", "test:multimodal-rag": "pnpm exec tsx test/continuous-test-suite-multimodal-rag.ts", "test:multimodal-text-content-fold": "pnpm exec tsx test/continuous-test-suite-multimodal-text-content-fold.ts", "test:proxy-sharing": "pnpm run build && pnpm exec tsx test/continuous-test-suite-proxy-sharing.ts", diff --git a/src/lib/neurolink.ts b/src/lib/neurolink.ts index f45eeb176..c7b711a29 100644 --- a/src/lib/neurolink.ts +++ b/src/lib/neurolink.ts @@ -32,6 +32,7 @@ import type { AgentRunOutcome, AgentToolRegistrationOptions, IsolatedAgentDefinition, + LogEventEmitter, NetworkExecutionInput, NetworkExecutionOptions, NetworkExecutionResult, @@ -687,6 +688,9 @@ function createTypedEmitter< return emitter as TypedEventEmitter; } +/** Monotonic source for {@link NeuroLink.logInstanceId}. Process-local by design. */ +let neuroLinkInstanceCounter = 0; + export class NeuroLink { /** @internal Brand for cross-module identification — see {@link isNeuroLink}. */ readonly [NEUROLINK_BRAND] = true as const; @@ -696,6 +700,31 @@ export class NeuroLink { private mcpInitPromise: Promise | null = null; private emitter = createTypedEmitter(); + /** + * Process-unique id used to attribute this instance's log events. The SDK + * entry points run their bodies inside `logger.runInInstanceScope(this.logInstanceId, …)`, + * which is what lets a worker's log bridge receive only the worker's own + * events instead of everything in the process. + */ + private readonly logInstanceId = `nl-${++neuroLinkInstanceCounter}`; + + /** + * Detach callbacks for scoped log bridges of workers this instance created + * via {@link createWorkerInstance} with `onLog`, that have not yet been + * removed by the worker's own `dispose()`. The scoped-emitter registry + * lives on the process-global logger, keyed by the WORKER's id — not this + * host's — so it survives this host being garbage collected. If a caller + * disposes the host but forgets an individual worker (the host is the + * long-lived object in most delegation patterns; the worker the + * short-lived one), the bridge — and everything its `onLog` closes over — + * would otherwise stay registered for the life of the process. `dispose()` + * sweeps whatever is still here so host cleanup is sufficient on its own, + * matching the pre-scoped-routing behavior where a worker's forward + * listener lived on the host's own emitter and was removed by + * `this.emitter.removeAllListeners()`. + */ + private readonly ownedWorkerLogBridgeDetachers = new Set<() => void>(); + // TaskManager — lazy-initialized on first access via `this.tasks` private _taskManager?: TaskManager; private _taskManagerConfig?: TaskManagerConfig; @@ -4463,6 +4492,18 @@ Current user's request: ${currentInput}`; */ async generate( optionsOrPrompt: GenerateOptions | DynamicOptions | string, + ): Promise { + return logger.runInInstanceScope(this.logInstanceId, () => + this.generateInInstanceScope(optionsOrPrompt), + ); + } + + /** + * Body of {@link generate}, always entered through the instance log scope + * so every log call beneath it is attributable to this instance. + */ + private async generateInInstanceScope( + optionsOrPrompt: GenerateOptions | DynamicOptions | string, ): Promise { // Host-loop delegation (registerAgentTool): enter a per-turn scope so // delegation caps count against THIS top-level generate, and withhold @@ -6693,6 +6734,17 @@ Current user's request: ${currentInput}`; */ async generateText( options: TextGenerationOptions, + ): Promise { + return logger.runInInstanceScope(this.logInstanceId, () => + this.generateTextInInstanceScope(options), + ); + } + + /** + * Body of {@link generateText}, always entered through the instance log scope. + */ + private async generateTextInInstanceScope( + options: TextGenerationOptions, ): Promise { // Validate required parameters for backward compatibility if ( @@ -9332,6 +9384,23 @@ Current user's request: ${currentInput}`; * @throws {Error} When conversation memory operations fail (if enabled) */ async stream(options: StreamOptions | DynamicOptions): Promise { + return logger.runInInstanceScope(this.logInstanceId, () => + this.streamInInstanceScope(options), + ); + } + + /** + * Body of {@link stream}, always entered through the instance log scope. + * + * Note the scope covers setting the stream up, not draining it: logs + * emitted while the consumer iterates the returned stream run in the + * consumer's async context, so they are attributed to whatever scope the + * consumer is in. Provider loops that run to completion inside this call + * are covered. + */ + private async streamInInstanceScope( + options: StreamOptions | DynamicOptions, + ): Promise { // Host-loop delegation (registerAgentTool): the same per-turn scope as // generate() — delegation caps count against THIS streamed turn and // depth-limited agent tools are withheld from the request. The provider @@ -18252,10 +18321,32 @@ Current user's request: ${currentInput}`; // Log-bridge listener errors never disrupt the worker. } }; - hostEmitter.on("log-event", forward); + // Bridge the WORKER's own scoped sink, not the host's process-wide one. + // Subscribing to hostEmitter would deliver every log event in the + // process — host, sibling workers, MCP — stamped with this worker's + // tag. The scoped registration delivers only events logged inside the + // worker's own generate/stream/generateText calls. + const bridgeEmitter: LogEventEmitter = { + emit: (event: string, ...args: unknown[]) => { + if (event === "log-event") { + forward(args[0]); + } + return true; + }, + }; + const workerLogInstanceId = worker.logInstanceId; + logger.addScopedEventEmitter(workerLogInstanceId, bridgeEmitter); + // Tracked on the HOST so that disposing the host — not just the + // worker — is enough to remove this bridge if the caller never + // disposes the worker itself. See `ownedWorkerLogBridgeDetachers`. + const detachBridge = (): void => { + logger.removeScopedEventEmitter(workerLogInstanceId, bridgeEmitter); + this.ownedWorkerLogBridgeDetachers.delete(detachBridge); + }; + this.ownedWorkerLogBridgeDetachers.add(detachBridge); const originalDispose = worker.dispose.bind(worker); worker.dispose = async () => { - hostEmitter.off("log-event", forward); + detachBridge(); await originalDispose(); }; } @@ -19050,6 +19141,10 @@ Current user's request: ${currentInput}`; // Clear only if this instance's emitter is the active log sink — // disposing a worker instance must not yank the host's bridge. logger.clearEventEmitter(this.emitter); + // Scoped sinks live on the process-global logger, keyed by this + // instance's id — they outlive `this.emitter` and must be dropped + // explicitly or the sink (and its closure) leaks for the process. + logger.clearScopedEventEmitters(this.logInstanceId); logger.debug("[NeuroLink] Event listeners removed successfully"); } catch (error) { const err = @@ -19061,6 +19156,29 @@ Current user's request: ${currentInput}`; } } + // 4a. Detach any worker log bridges this instance created (via + // createWorkerInstance({ onLog })) that the caller never disposed + // individually. Their scoped sink lives on the process-global logger + // keyed by the WORKER's id, so it would otherwise outlive this host. + if (this.ownedWorkerLogBridgeDetachers.size > 0) { + try { + const count = this.ownedWorkerLogBridgeDetachers.size; + for (const detach of [...this.ownedWorkerLogBridgeDetachers]) { + detach(); + } + logger.debug( + `[NeuroLink] Detached ${count} undisposed worker log bridge(s)`, + ); + } catch (error) { + const err = + error instanceof Error + ? error + : new Error(`Worker log bridge cleanup error: ${String(error)}`); + cleanupErrors.push(err); + logger.warn("[NeuroLink] Error detaching worker log bridges:", error); + } + } + // 5. Clear all circuit breakers if (this.toolCircuitBreakers && this.toolCircuitBreakers.size > 0) { try { diff --git a/src/lib/types/isolatedAgent.ts b/src/lib/types/isolatedAgent.ts index 5b9aeac28..5b89c86ce 100644 --- a/src/lib/types/isolatedAgent.ts +++ b/src/lib/types/isolatedAgent.ts @@ -55,13 +55,22 @@ export type WorkerInstanceOptions = { * Log bridge sink. Fire-and-forget: listener errors never disrupt the * worker. * - * KNOWN LIMITATION: the underlying NeuroLink logger is process-global with - * a single active emitter, so this bridge receives ALL NeuroLink log - * events emitted in the process (host, other workers, MCP) for the - * worker's lifetime, each stamped with this worker's `tag`. The tag - * identifies which bridge forwarded the event, not which instance emitted - * it. Per-instance log attribution requires per-instance logger routing — - * tracked as follow-up work in the RFC. + * Receives only log events this worker emitted. The logger routes per + * instance: the worker's `generate` / `stream` / `generateText` run their + * bodies inside an AsyncLocalStorage scope carrying the worker's id, and + * this bridge is subscribed to that id — so a sibling worker's, the host's + * or a background MCP reconnect's logs never arrive here, and `tag` is + * true attribution rather than "whichever bridge forwarded it". + * + * Two things stay outside the scope, by construction: + * - Logs emitted while a **consumer drains a returned stream**. Iteration + * happens in the consumer's async context, after `stream()` resolved. + * - Logs emitted **outside any call** — construction, background MCP + * reconnects, module init. These are unattributed rather than charged to + * an arbitrary instance. + * + * A process-wide sink (`logger.setEventEmitter`) still sees everything, so + * a host bridge is unaffected by this narrowing. */ onLog?: (event: WorkerLogEvent) => void; /** diff --git a/src/lib/types/utilities.ts b/src/lib/types/utilities.ts index 7be78b578..a6b81bdd3 100644 --- a/src/lib/types/utilities.ts +++ b/src/lib/types/utilities.ts @@ -74,9 +74,25 @@ export type LogEntry = { data?: unknown; }; +/** + * Minimal emitter shape the logger forwards `"log-event"` payloads to. + * + * Structurally satisfied by `node:events`' `EventEmitter` and by NeuroLink's + * own typed emitter, so a sink can be either. + */ +export type LogEventEmitter = { + emit: (event: string, ...args: unknown[]) => boolean; +}; + /** * Logger interface matching the logger object shape * Used for SDK tool contexts and other components that need a logger + * + * Deliberately a subset of the real logger: the per-instance routing methods + * (`runInInstanceScope`, `addScopedEventEmitter`, …) are internal plumbing + * between the SDK entry points and the logger, and this type is a structural + * contract a caller can satisfy — `SDKToolContext.logger`. Adding required + * members here would break anyone constructing that context themselves. */ export type Logger = { debug: (...args: unknown[]) => void; @@ -88,9 +104,7 @@ export type Logger = { setLogLevel: (level: LogLevel) => void; getLogs: (level?: LogLevel) => LogEntry[]; clearLogs: () => void; - setEventEmitter: (emitter: { - emit: (event: string, ...args: unknown[]) => boolean; - }) => void; + setEventEmitter: (emitter: LogEventEmitter) => void; clearEventEmitter: () => void; }; diff --git a/src/lib/utils/logger.ts b/src/lib/utils/logger.ts index 4f91ab336..d313afd34 100644 --- a/src/lib/utils/logger.ts +++ b/src/lib/utils/logger.ts @@ -14,7 +14,21 @@ * - Tabular data display */ -import type { LogEntry, LogLevel } from "../types/index.js"; +import { AsyncLocalStorage } from "node:async_hooks"; +import type { LogEntry, LogEventEmitter, LogLevel } from "../types/index.js"; + +/** + * Identifies which NeuroLink instance is emitting, for the async operation + * currently on the stack. Populated by `runInInstanceScope()` — which the SDK + * entry points (`generate`, `stream`, `generateText`) wrap their bodies in — + * so a log call anywhere beneath them can be attributed without every call + * site threading an instance through. + * + * Empty for logs emitted outside any call (construction, background MCP + * reconnects, module init). Those stay unattributed rather than being charged + * to whichever instance happens to hold the global sink. + */ +const instanceLogScope = new AsyncLocalStorage(); // OTel trace context for log correlation (optional — gracefully no-ops if OTel not initialized) let traceApi: typeof import("@opentelemetry/api") | null = null; @@ -64,15 +78,38 @@ function argvRequestsJsonOutput(): boolean { ); } +/** + * Best-effort isolated copy of a log call's `data` payload for one consumer. + * `NeuroLinkLogger.log()` hands the same call's data to every scoped emitter + * for the active instance plus the process-wide sink — a designed, supported + * multi-consumer configuration — so a copy per consumer is what keeps one + * consumer's in-place mutation (e.g. redacting a field before forwarding it) + * from silently corrupting what a sibling consumer, or stored log history, + * observes. Non-object values need no copy — they are already by-value. + * Values `structuredClone` cannot handle (functions, etc.) are handed through + * unchanged rather than dropped: isolation is not achievable for those + * without lossy serialization, so that narrow case keeps the old + * shared-reference behavior instead of losing the payload. + */ +function cloneLogData(data: unknown): unknown { + if (data === null || typeof data !== "object") { + return data; + } + try { + return structuredClone(data); + } catch { + return data; + } +} + class NeuroLinkLogger { private logLevel: LogLevel = "info"; private logs: LogEntry[] = []; private maxLogs = 1000; private isDebugMode: boolean; private diagnosticsToStderr = false; - private eventEmitter?: { - emit: (event: string, ...args: unknown[]) => boolean; - }; + private eventEmitter?: LogEventEmitter; + private readonly scopedEmitters = new Map>(); constructor() { // Cache debug mode check to avoid repeated array searches @@ -94,9 +131,7 @@ class NeuroLinkLogger { * * @param emitter - The event emitter instance */ - setEventEmitter(emitter: { - emit: (event: string, ...args: unknown[]) => boolean; - }): void { + setEventEmitter(emitter: LogEventEmitter): void { this.eventEmitter = emitter; } @@ -109,15 +144,87 @@ class NeuroLinkLogger { * * @param ifEmitter - When provided, clear only if it is the current emitter */ - clearEventEmitter(ifEmitter?: { - emit: (event: string, ...args: unknown[]) => boolean; - }): void { + clearEventEmitter(ifEmitter?: LogEventEmitter): void { if (ifEmitter !== undefined && this.eventEmitter !== ifEmitter) { return; } this.eventEmitter = undefined; } + /** + * Runs `fn` with every log call beneath it attributed to `instanceId`. + * + * Nesting is safe and expected: a host turn that delegates to a worker ends + * up with the worker's id on top for the duration of the worker's call, and + * the host's id restored afterwards. Re-entering with the same id is a + * no-op in effect. + * + * @param instanceId - Identifier of the emitting NeuroLink instance + * @param fn - Work to run inside the scope + * @returns Whatever `fn` returns + */ + runInInstanceScope(instanceId: string, fn: () => T): T { + return instanceLogScope.run(instanceId, fn); + } + + /** + * The instance id currently attributed, or undefined outside any scope. + */ + getInstanceScope(): string | undefined { + return instanceLogScope.getStore(); + } + + /** + * Subscribes `emitter` to log events emitted by one instance only. + * + * Unlike {@link setEventEmitter} — a single process-wide sink that receives + * everything — a scoped emitter receives only events logged inside that + * instance's {@link runInInstanceScope}. Several emitters may share an id. + * + * @param instanceId - Instance whose events this emitter should receive + * @param emitter - The sink to subscribe + */ + addScopedEventEmitter(instanceId: string, emitter: LogEventEmitter): void { + const existing = this.scopedEmitters.get(instanceId); + if (existing) { + existing.add(emitter); + return; + } + this.scopedEmitters.set(instanceId, new Set([emitter])); + } + + /** + * Unsubscribes a scoped emitter. Must be called when the owning instance is + * disposed, or the emitter — and everything it closes over — is retained by + * the process-global logger for the lifetime of the process. + * + * @param instanceId - Instance the emitter was registered against + * @param emitter - The sink to unsubscribe + */ + removeScopedEventEmitter(instanceId: string, emitter: LogEventEmitter): void { + const existing = this.scopedEmitters.get(instanceId); + if (!existing) { + return; + } + existing.delete(emitter); + if (existing.size === 0) { + this.scopedEmitters.delete(instanceId); + } + } + + /** + * Drops every scoped emitter registered against an instance. + * + * Called from `NeuroLink.dispose()`: the registry lives on the + * process-global logger, so an instance that goes away without clearing its + * own entry would keep its sinks — and their closures — reachable forever. + * + * @param instanceId - Instance whose sinks should be dropped + */ + clearScopedEventEmitters(instanceId: string): void { + this.scopedEmitters.delete(instanceId); + } + /** * Sets the minimum log level that will be processed and output. * Log messages with a level lower than this will be ignored. @@ -321,17 +428,49 @@ class NeuroLinkLogger { data, }; - // Emit log event if emitter is configured - if (this.eventEmitter) { - try { - this.eventEmitter.emit("log-event", { - level, - message, - timestamp: new Date().getTime(), - data, - }); - } catch { - // Silently ignore emitter errors to avoid disrupting logging + // Emit to the per-instance sinks for whichever instance is on the stack, + // then to the process-wide sink. The two are independent: a scoped sink + // never sees another instance's events, and the global sink still sees + // everything, so existing setEventEmitter() consumers are unaffected. + // + // Several emitters may be registered for the same instance id (see + // addScopedEventEmitter's own JSDoc), and the global sink always runs + // alongside them — so one call's `data` object is, by design, handed to + // more than one consumer. Each consumer below gets its OWN clone rather + // than the shared `data` reference, so an in-place mutation by one (e.g. + // a bridge redacting a field before forwarding it) cannot corrupt what a + // sibling consumer sees or what `entry` — which keeps the original, + // unmodified `data` — stores for `getLogs()`. + const scopeId = instanceLogScope.getStore(); + const scoped = + scopeId === undefined ? undefined : this.scopedEmitters.get(scopeId); + if (scoped || this.eventEmitter) { + const timestampMs = entry.timestamp.getTime(); + if (scoped) { + for (const emitter of scoped) { + try { + emitter.emit("log-event", { + level, + message, + timestamp: timestampMs, + data: cloneLogData(data), + }); + } catch { + // Silently ignore emitter errors to avoid disrupting logging + } + } + } + if (this.eventEmitter) { + try { + this.eventEmitter.emit("log-event", { + level, + message, + timestamp: timestampMs, + data: cloneLogData(data), + }); + } catch { + // Silently ignore emitter errors to avoid disrupting logging + } } } @@ -568,12 +707,20 @@ export const logger = { neuroLinkLogger.setDiagnosticsToStderr(enabled), getLogs: (level?: LogLevel) => neuroLinkLogger.getLogs(level), clearLogs: () => neuroLinkLogger.clearLogs(), - setEventEmitter: (emitter: { - emit: (event: string, ...args: unknown[]) => boolean; - }) => neuroLinkLogger.setEventEmitter(emitter), - clearEventEmitter: (ifEmitter?: { - emit: (event: string, ...args: unknown[]) => boolean; - }) => neuroLinkLogger.clearEventEmitter(ifEmitter), + setEventEmitter: (emitter: LogEventEmitter) => + neuroLinkLogger.setEventEmitter(emitter), + clearEventEmitter: (ifEmitter?: LogEventEmitter) => + neuroLinkLogger.clearEventEmitter(ifEmitter), + // Per-instance routing (see NeuroLinkLogger.runInInstanceScope) + runInInstanceScope: (instanceId: string, fn: () => T) => + neuroLinkLogger.runInInstanceScope(instanceId, fn), + getInstanceScope: () => neuroLinkLogger.getInstanceScope(), + addScopedEventEmitter: (instanceId: string, emitter: LogEventEmitter) => + neuroLinkLogger.addScopedEventEmitter(instanceId, emitter), + removeScopedEventEmitter: (instanceId: string, emitter: LogEventEmitter) => + neuroLinkLogger.removeScopedEventEmitter(instanceId, emitter), + clearScopedEventEmitters: (instanceId: string) => + neuroLinkLogger.clearScopedEventEmitters(instanceId), }; /** diff --git a/test/continuous-test-suite-logger-instance-routing.ts b/test/continuous-test-suite-logger-instance-routing.ts new file mode 100644 index 000000000..ca5b2d9d2 --- /dev/null +++ b/test/continuous-test-suite-logger-instance-routing.ts @@ -0,0 +1,431 @@ +#!/usr/bin/env tsx +/** + * Continuous Test Suite: per-instance logger routing (#1236), driven through + * the public surface (Rule 15). + * + * The NeuroLink logger is a process-global singleton. Before #1236 it had a + * single active sink, so a worker's `onLog` bridge — which subscribed to the + * host's emitter — received every log event in the process (host, sibling + * workers, MCP), each stamped with that worker's `logTag`. The tag said which + * bridge forwarded an event, not which instance emitted it. Measured against + * the pre-fix build, one worker's bridge saw 708 events where 299 were its + * own; these cases fail on that build and pass on this one. + * + * Everything here comes from `../dist/index.js`: `NeuroLink` for the + * instances under test, and the exported `logger` for its log history and + * for the process-wide sink. Taking the logger from `src/lib/` while the code + * under test logs through `dist`'s copy would watch a different singleton and + * pass silently. + * + * No credentials and no network: each instance is exercised with a provider + * name that is not registered, which throws inside `generate()` — after the + * SDK has already emitted a few hundred debug events on the way there, which + * is exactly the traffic whose attribution is under test. `NEUROLINK_DEBUG` + * has to be on for anything below `error` to be emitted at all, and console + * output is muted while the calls run so the suite's own output stays + * readable. + * + * Every negative assertion ("this bridge received nothing") is paired with a + * precondition proving logging actually happened on that call — otherwise + * "nothing arrived" would pass on a run where nothing ran. The precondition + * is `logger.getLogs()`, which records entries regardless of which sinks are + * attached. It is deliberately NOT a process-wide sink: installing one + * displaces whatever emitter is currently active, which would mask the very + * behaviour under test. The one case that does install a sink scopes it to + * itself and detaches it again. + * + * Run: npx tsx test/continuous-test-suite-logger-instance-routing.ts + */ +import { defineSuite, assert } from "./helpers/harness.js"; +import { NeuroLink, logger } from "../dist/index.js"; + +const { test, runSuite } = defineSuite("Per-instance logger routing (#1236)", { + offline: true, +}); + +/** An unregistered provider name: rejects inside generate(), reaches no network. */ +const UNREGISTERED_PROVIDER = "not-a-registered-provider-1236"; + +type Captured = { tag: string; level: string; message: string }; + +/** + * Drive one instance until it throws, discarding the rejection, and report + * how many entries the logger recorded while it ran. + * + * The count is this suite's precondition: a negative assertion about routing + * means nothing on a call that logged nothing. `clearLogs()` first so the + * count is this call's, not the run's — the logger keeps only its last 1000 + * entries and a single call emits several hundred. + * + * Console output is muted for the duration: with NEUROLINK_DEBUG on, one call + * prints hundreds of lines and would bury the suite's own results. The mute + * is restored in a finally so a throw cannot leave the process mute. + */ +async function emitLogsFrom(instance: NeuroLink): Promise { + const original = { + log: console.log, + info: console.info, + warn: console.warn, + error: console.error, + debug: console.debug, + }; + const mute = () => {}; + logger.clearLogs(); + console.log = mute; + console.info = mute; + console.warn = mute; + console.error = mute; + console.debug = mute; + try { + await instance.generate({ + input: { text: "per-instance logger routing probe" }, + provider: UNREGISTERED_PROVIDER, + timeout: 5_000, + }); + } catch { + // Expected: the provider name is not registered. The logs emitted on the + // way to that rejection are the subject of this suite. + } finally { + console.log = original.log; + console.info = original.info; + console.warn = original.warn; + console.error = original.error; + console.debug = original.debug; + } + return logger.getLogs().length; +} + +const originalDebugEnv = process.env.NEUROLINK_DEBUG; +const originalLevelEnv = process.env.NEUROLINK_LOG_LEVEL; +process.env.NEUROLINK_DEBUG = "true"; +process.env.NEUROLINK_LOG_LEVEL = "debug"; + +try { + const host = new NeuroLink(); + const capturedA: Captured[] = []; + const capturedB: Captured[] = []; + + // createWorkerInstance registers worker A's bridge under an instance id + // nothing outside NeuroLink can name (it is a private field). Spying on + // the logger's own public registration hook for the duration of this one + // synchronous call captures the real id, so the dispose test below can + // re-enter worker A's own scope directly afterwards, rather than relying + // on worker B's scope — which can never reach A's emitter regardless of + // whether dispose removed it. + let workerAInstanceId: string | undefined; + const originalAddScopedEventEmitter = logger.addScopedEventEmitter; + logger.addScopedEventEmitter = (instanceId, emitter) => { + workerAInstanceId = instanceId; + originalAddScopedEventEmitter(instanceId, emitter); + }; + const workerA = host.createWorkerInstance({ + logTag: "worker-a", + onLog: (event) => + capturedA.push({ + tag: event.tag, + level: event.level, + message: event.message, + }), + }); + logger.addScopedEventEmitter = originalAddScopedEventEmitter; + + const workerB = host.createWorkerInstance({ + logTag: "worker-b", + onLog: (event) => + capturedB.push({ + tag: event.tag, + level: event.level, + message: event.message, + }), + }); + + await test("a worker's bridge receives that worker's own log events", async () => { + const emitted = await emitLogsFrom(workerA); + + assert(emitted > 0, "precondition: the call recorded log entries"); + assert( + capturedA.length > 0, + `worker A's bridge received none of its own events (got ${capturedA.length})`, + ); + }); + + await test("a sibling worker's bridge receives none of them", async () => { + assert( + capturedA.length > 0, + "precondition: worker A's bridge captured events from the previous call", + ); + assert( + capturedB.length === 0, + `worker B's bridge received events from a call it did not make (got ${capturedB.length})`, + ); + }); + + await test("the host's own call reaches neither worker bridge", async () => { + const aBefore = capturedA.length; + const bBefore = capturedB.length; + + const emitted = await emitLogsFrom(host); + + assert(emitted > 0, "precondition: the host call recorded log entries"); + assert( + capturedA.length === aBefore, + `worker A's bridge grew on a host call (by ${capturedA.length - aBefore})`, + ); + assert( + capturedB.length === bBefore, + `worker B's bridge grew on a host call (by ${capturedB.length - bBefore})`, + ); + }); + + await test("routing is reciprocal: B's call reaches B and not A", async () => { + const aBefore = capturedA.length; + + const emitted = await emitLogsFrom(workerB); + + assert(emitted > 0, "precondition: the call recorded log entries"); + assert( + capturedB.length > 0, + `worker B's bridge received none of its own events (got ${capturedB.length})`, + ); + assert( + capturedA.length === aBefore, + `worker A's bridge grew on worker B's call (by ${capturedA.length - aBefore})`, + ); + }); + + await test("every forwarded event carries the owning bridge's tag", async () => { + assert( + capturedA.length > 0 && capturedB.length > 0, + "precondition: both bridges captured events", + ); + const strayA = capturedA.filter((e) => e.tag !== "worker-a").length; + const strayB = capturedB.filter((e) => e.tag !== "worker-b").length; + assert(strayA === 0, `worker A's bridge stamped ${strayA} wrong tags`); + assert(strayB === 0, `worker B's bridge stamped ${strayB} wrong tags`); + }); + + await test("a process-wide sink still receives every instance's events", async () => { + // Backward compatibility: narrowing the worker bridges must not narrow + // logger.setEventEmitter(), which existing hosts rely on. Installing one + // also displaces whichever instance emitter is active, so this case + // attaches and detaches it rather than leaving it in place. + let seen = 0; + const sink = { + emit: (event: string) => { + if (event === "log-event") { + seen++; + } + return true; + }, + }; + logger.setEventEmitter(sink); + try { + const fromHost = await emitLogsFrom(host); + const afterHost = seen; + const fromWorker = await emitLogsFrom(workerB); + + assert( + fromHost > 0 && fromWorker > 0, + "precondition: both calls recorded log entries", + ); + assert( + afterHost > 0, + `the process-wide sink missed the host's events (got ${afterHost})`, + ); + assert( + seen > afterHost, + `the process-wide sink missed a worker's events (grew by ${seen - afterHost})`, + ); + } finally { + logger.clearEventEmitter(sink); + } + }); + + await test("a scoped emitter's in-place mutation of data does not corrupt the global sink or log history", async () => { + // addScopedEventEmitter's own JSDoc documents several emitters sharing an + // id, and the global sink runs alongside every scoped one on the SAME + // call — a designed multi-consumer configuration. A worker bridge doing + // an ordinary thing (redacting a field before it forwards an event) + // must not be able to reach into what a sibling consumer, or stored log + // history, observes for that same call. + const probeInstanceId = "shared-mutable-data-probe-1236"; + let globalSawSecret: boolean | undefined; + const globalSink = { + emit: (event: string, ...args: unknown[]) => { + if (event === "log-event") { + const ev = args[0] as { data?: { secret?: unknown } } | undefined; + globalSawSecret = ev?.data?.secret === "sk-probe"; + } + return true; + }, + }; + const redactingEmitter = { + emit: (event: string, ...args: unknown[]) => { + if (event === "log-event") { + const ev = args[0] as { data?: Record } | undefined; + if (ev?.data && typeof ev.data === "object") { + delete ev.data.secret; + } + } + return true; + }, + }; + + logger.clearLogs(); + logger.setEventEmitter(globalSink); + logger.addScopedEventEmitter(probeInstanceId, redactingEmitter); + try { + // error, like the other explicit single-call probes in this suite: + // shouldLog() gates "debug" on `this.logLevel`, which the logger + // singleton reads from NEUROLINK_LOG_LEVEL only once, at module import + // — before this file's top-level `process.env.NEUROLINK_LOG_LEVEL = + // "debug"` assignment runs. "error" always passes shouldLog() + // regardless of that timing, which is exactly why the suite's other + // direct probes use it too. + logger.runInInstanceScope(probeInstanceId, () => { + logger.error("[#1236] shared-data probe", { + secret: "sk-probe", + keep: true, + }); + }); + + assert( + globalSawSecret !== undefined, + "precondition: the global sink received the probe event", + ); + assert( + globalSawSecret === true, + "a sibling scoped emitter's in-place redaction leaked into the global sink's data", + ); + + const lastEntry = logger.getLogs()[logger.getLogs().length - 1]; + const storedData = lastEntry?.data as Record | undefined; + assert( + storedData?.secret === "sk-probe", + "a sibling scoped emitter's in-place redaction corrupted stored log history", + ); + } finally { + logger.removeScopedEventEmitter(probeInstanceId, redactingEmitter); + logger.clearEventEmitter(globalSink); + } + }); + + await test("dispose() unsubscribes a worker's bridge", async () => { + assert( + workerAInstanceId !== undefined, + "precondition: worker A's construction registered a scoped emitter", + ); + + await workerA.dispose(); + const aAfterDispose = capturedA.length; + + const emitted = await emitLogsFrom(workerB); + + assert( + emitted > 0, + "precondition: the post-dispose call recorded log entries", + ); + assert( + capturedA.length === aAfterDispose, + `a disposed worker's bridge still received events (${capturedA.length - aAfterDispose} after dispose)`, + ); + + // The two assertions above hold whether or not dispose() actually + // unsubscribed anything: worker B's own AsyncLocalStorage scope can + // never reach worker A's scoped emitter, so a retained bridge would + // slip through undetected. Re-enter worker A's own scope directly + // (the id captured at construction) and prove the removal for real. + logger.runInInstanceScope(workerAInstanceId as string, () => { + logger.error("[#1236] dispose-removal probe"); + }); + assert( + capturedA.length === aAfterDispose, + "worker A's bridge still received an event when its own scope was re-entered after dispose", + ); + }); + + await test("disposing the host also removes a still-undisposed worker's log bridge", async () => { + // The scoped-emitter registry lives on the process-global logger, keyed + // by the WORKER's id, not the host's — so a caller who disposes only the + // long-lived host and forgets the short-lived worker must not leak that + // worker's bridge (and everything its onLog closes over) for the rest of + // the process. A separate host/worker pair keeps this from disturbing + // the suite's shared host/workerA/workerB above. + const scopedHost = new NeuroLink(); + let scopedWorkerInstanceId: string | undefined; + const originalAddScopedEventEmitter = logger.addScopedEventEmitter; + logger.addScopedEventEmitter = (instanceId, emitter) => { + scopedWorkerInstanceId = instanceId; + originalAddScopedEventEmitter(instanceId, emitter); + }; + const capturedScoped: Captured[] = []; + const scopedWorker = scopedHost.createWorkerInstance({ + logTag: "host-dispose-sweep-probe", + onLog: (event) => + capturedScoped.push({ + tag: event.tag, + level: event.level, + message: event.message, + }), + }); + logger.addScopedEventEmitter = originalAddScopedEventEmitter; + + assert( + scopedWorkerInstanceId !== undefined, + "precondition: the worker's construction registered a scoped emitter", + ); + + // The caller disposes only the host — the worker itself is never + // disposed. This is the shape of the leak: forgetting one cleanup call + // in a delegation pattern where the host outlives the worker. + await scopedHost.dispose(); + + logger.runInInstanceScope(scopedWorkerInstanceId as string, () => { + logger.error("[#1236] host-dispose sweep probe"); + }); + assert( + capturedScoped.length === 0, + `the worker's bridge still received an event after only the host was disposed (got ${capturedScoped.length})`, + ); + + await scopedWorker.dispose(); + }); + + await test("logs outside any generate call are attributed to no instance", async () => { + // Construction, background reconnects and module init sit outside every + // entry point's scope, so they must not be charged to an arbitrary + // instance. getInstanceScope() reports the scope at the call site. + assert( + logger.getInstanceScope() === undefined, + "a log emitted from suite top level reported an owning instance", + ); + + const bBefore = capturedB.length; + logger.clearLogs(); + logger.error("[#1236] unattributed probe emitted outside any entry point"); + assert( + logger.getLogs().length > 0, + "precondition: the unscoped log was recorded", + ); + assert( + capturedB.length === bBefore, + `an unscoped log reached a worker bridge (by ${capturedB.length - bBefore})`, + ); + }); + + await workerB.dispose(); + await host.dispose(); +} finally { + if (originalDebugEnv === undefined) { + delete process.env.NEUROLINK_DEBUG; + } else { + process.env.NEUROLINK_DEBUG = originalDebugEnv; + } + if (originalLevelEnv === undefined) { + delete process.env.NEUROLINK_LOG_LEVEL; + } else { + process.env.NEUROLINK_LOG_LEVEL = originalLevelEnv; + } +} + +await runSuite();