Skillpacks that bring in data

01How a skillpack source works

A skillpack that brings in data is a folder with a manifest file named nexusbrain.plugin.json at its root. The manifest has an ingestion_sources list. Each entry points at a JavaScript or TypeScript module whose default export is a factory function: nexusbrain calls it with a config object and gets back a source.

nexusbrain finds skillpack folders through the NEXUS_BRAIN_PLUGIN_PATH environment variable. This is the same variable that plugin subagent definitions use, so one folder can ship both.

Once running, a source only does one thing: it calls ctx.emit(event) with each new piece of content. nexusbrain validates the event, drops duplicates, applies a rate limit and queues it for writing. A source never writes to the brain directly.

02Make a skillpack visible to nexusbrain

1. 1

   Put the skillpack in a folder with an absolute path.

   The folder must contain nexusbrain.plugin.json at its top level and the module file named in module, for example ./dist/source.js.

2. 2

   Check the manifest.

   It should look like this. The kind, module and default_config values are examples.

   {
     "name": "granola-source",
     "plugin_version": "nexusbrain-plugin-v1",
     "ingestion_sources": [
       {
         "kind": "voice-granola",
         "module": "./dist/source.js",
         "api_version": "nexusbrain-ingestion-source-v1",
         "default_config": { "transcription_model": "whisper-1" },
         "permissions": ["network"]
       }
     ]
   }

   Copy
3. 3

   Add the folder to NEXUS_BRAIN_PLUGIN_PATH.

   The value is a list of absolute paths separated by colons, like PATH. Relative paths, paths starting with ~, and URLs are rejected. When two skillpacks declare the same kind, the one listed first wins.

   export NEXUS_BRAIN_PLUGIN_PATH="/home/me/packs/granola-source:/home/me/packs/other-pack"

   Copy

   You should seeNothing is printed. The variable is read the next time nexusbrain loads skillpack sources.

03Build your own source

The contract is exported from nexusbrain/ingestion. Treat it as a versioned public API: IngestionSource, IngestionEvent, IngestionSourceContext, computeContentHash, validateIngestionEvent and the INGESTION_SOURCE_API_VERSION constant.

1. 1

   Write the factory as the default export.

   The factory receives the merged config and returns an object with id, kind, start and stop. healthCheck is optional. start must resolve once the source is ready to emit. stop must finish in-flight work quickly; long waits should also listen to ctx.abortSignal.

   import type { IngestionSource, IngestionSourceContext } from 'nexusbrain/ingestion';
   import { computeContentHash } from 'nexusbrain/ingestion';
   
   export default function createSource(config: Record<string, unknown>): IngestionSource {
     let ctx: IngestionSourceContext | null = null;
     return {
       id: 'voice-granola-main',
       kind: 'voice-granola',
       async start(c) {
         ctx = c;
         const text = '# Meeting notes\n...';
         ctx.emit({
           source_id: 'voice-granola-main',
           source_kind: 'voice-granola',
           source_uri: 'granola://meeting/123',
           received_at: new Date().toISOString(),
           content_type: 'text/markdown',
           content: text,
           content_hash: computeContentHash(text),
         });
       },
       async stop() {
         ctx = null;
       },
     };
   }

   Copy
2. 2

   Fill in every required event field.

   Events that fail validation are dropped and logged, and the source keeps running. The rules come straight from the validator.

3. 3

   Throw when something unrecoverable happens.

   If start throws, nexusbrain restarts the source with exponential backoff, starting at 1 second and capped at 60 seconds. After 10 consecutive crashes the source is marked failed and not retried. A source that runs for 5 minutes before crashing has its crash count reset.

04Test a source without a brain

The package path nexusbrain/ingestion/test-harness gives you IngestionTestHarness and expectEvent. The harness passes the same validation as the real daemon but records events instead of writing them.

1. 1

   Write a unit test that runs the source in the harness.

   import { test, expect } from 'bun:test';
   import { IngestionTestHarness, expectEvent } from 'nexusbrain/ingestion/test-harness';
   import createSource from './source';
   
   test('emits one valid event', async () => {
     const harness = new IngestionTestHarness({ config: { transcription_model: 'whisper-1' } });
     await harness.run(createSource({}));
     expect(harness.validationErrors).toHaveLength(0);
     expect(harness.events).toHaveLength(1);
     expectEvent(harness.events[0]).toHaveKind('voice-granola');
     expectEvent(harness.events[0]).toHaveContentHash();
     await harness.stop();
   });

   Copy
2. 2

   Run the test.

   bun test

   Copy

   You should seeThe test passes. If an event was malformed, harness.validationErrors lists the field and the reason, for example content_hash: must be 64 lowercase hex characters (SHA-256).

• expectEvent(e) also offers toHaveSourceId, toHaveSourceUri, toHaveContentType, toBeUntrusted, toBeTrusted and toHaveMetadata.
• The harness gives the source no database. If the source reads ctx.engine, pass { engine } to the harness or the test fails with a message naming the property it tried to use.
• harness.stop() waits 5 seconds by default for the source to stop, then fails with source did not stop within 5000ms.

05Troubleshooting