Agent Stream

Async-iterable variants for agent methods that yield N rows. Operator wraps a cursor-shaped fetcher with b.agent.stream.create; the resulting object is AsyncIterable — built-in backpressure (each yield blocks until the consumer pulls), automatic cursor close via try/finally on any exit path (consumer break, throw, network drop), and drain-marker emit on orchestrator drain so clients can resume from lastSeenCursor against the new agent post-drain.

var stream = b.agent.stream.create({
  orchestrator: orch,                  // optional — for drain reg
  actor:        { id: "u1" },
  kind:         "search",
  batchSize:    256,
  openCursor:   function (cursorOpts) {
    return store.openSearchCursor(cursorOpts);     // operator
  },
  cursorOpts:   { folder: "INBOX", sinceModseq: 0 },
});

for await (var row of stream) {
  // row delivered as soon as the cursor yields it.
  // Pulling slowly applies backpressure to the store.
  if (someCondition) break;            // cursor.close() fires automatically
}

## Drain-marker semantic

When orchestrator drain fires, in-flight streams emit ONE final { _drainMarker: true, lastSeenCursor: , reason: "drain" } row and exit cleanly. Clients reconnecting via JMAP-WebSocket / IMAP NOTIFY pass lastSeenCursor back to resume.

## Cursor contract

Operator-supplied cursor: cursor.fetchBatch(batchSize) → { rows, nextCursor, done } cursor.close() → void | Promise

The framework's b.mailStore will gain openSearchCursor / openFolderCursor / openExportCursor etc. at later mail-stack slices that compose this primitive.

b.agent.stream.create(opts) #

{
  openCursor:    function(cursorOpts) → cursor,   // required
  cursorOpts:    object,                           // operator-passed
  batchSize:     integer,                           // default 256
  orchestrator:  b.agent.orchestrator,              // optional — for drain reg
  actor:         { id, ... },                       // optional — audit attribution
  kind:          string,                            // "search" / "export" / ...
  audit:         b.audit,                           // optional
}

Create an async-iterable backed by an operator-supplied cursor. Returns an object that implements [Symbol.asyncIterator] — usable with for await (var row of stream). Cursor close + audit emit + orchestrator stream-registry hook are owned by the framework; operator only supplies the openCursor factory + cursorOpts.

var stream = b.agent.stream.create({
  openCursor: function (o) { return store.openSearchCursor(o); },
  cursorOpts: { folder: "INBOX" },
});
for await (var row of stream) { process(row); }

Last updated 2026-08-08T16:39:49.652Z by seeder.