Skip to main content

Class: CrewSession

Defined in: packages/orchestration/src/crew/crew-session.ts:19

Implements​

  • Processable<string, string>

Constructors​

Constructor​

new CrewSession(coordinator, notificationBus, orchestrator, config): CrewSession

Defined in: packages/orchestration/src/crew/crew-session.ts:32

Parameters​

coordinator​

Baleybot

notificationBus​

NotificationBus

orchestrator​

Orchestrator

config​
onCoordinatorStreamEvent?​

(event, agentName) => void

Called for every streaming event from the coordinator agent. Enables real-time SSE streaming of the coordinator's response.

onWorkerStreamEvent?​

(runId, event, agentName) => void

Called for every streaming event from worker agents. Enables real-time UI updates (e.g., parsing canvas tags from agent output, showing thinking indicators, displaying tool progress).

proactiveIntervalMs​

number

proactiveUpdates​

boolean

Returns​

CrewSession

Properties​

events​

readonly events: EventBus<CrewEventMap>

Defined in: packages/orchestration/src/crew/crew-session.ts:20

Accessors​

isPaused​

Get Signature​

get isPaused(): boolean

Defined in: packages/orchestration/src/crew/crew-session.ts:176

Whether the crew is currently paused

Returns​

boolean

Methods​

close()​

close(): Promise<void>

Defined in: packages/orchestration/src/crew/crew-session.ts:289

Release what the session itself holds: the proactive-update heartbeat.

A setInterval is a host resource in the same sense a container is — it keeps the event loop alive, so a Node process that started a session and finished with it never exits. stop() clears it too, but stop() is a conversation ending: it asks the coordinator for a closing summary, which is a model call, and a teardown path must not make one.

Per the Processable.close() contract this is idempotent and does not end the session: on its own, send() still works afterwards, minus the unprompted status updates. Via Crew.dispose() it will not, but that is the crew's doing — the orchestrator is stopped and its event bus cleared — not this method's.

The coordinator is deliberately left alone: it may be one the caller supplied via CrewConfig.coordinator, and this node does not own it. Whoever built it closes it (Crew.dispose() does, for the ones it built).

Returns​

Promise<void>

Implementation of​

Processable.close

getBotNames()​

getBotNames(): string[]

Defined in: packages/orchestration/src/crew/crew-session.ts:249

Get all bot names in this processable (including nested processables)

Returns​

string[]

Array of bot names that can emit events

Example​

const pipeline = pipeline()
  .step(chatBot)
  .parallel({ sentiment: sentimentBot, tone: toneBot })
  .build();

const botNames = pipeline.getBotNames();
// ['assistant', 'sentiment', 'tone']

Implementation of​

Processable.getBotNames

getChildren()​

getChildren(): ProcessableChild[]

Defined in: packages/orchestration/src/crew/crew-session.ts:266

The coordinator, which is the one processable a session composes directly.

Worker agents are reached through the orchestrator at runtime rather than held here, so they are not children of this node — they are not fixed by the composition, and which ones run depends on what the coordinator decides to delegate.

Returns​

ProcessableChild[]

Implementation of​

Processable.getChildren

getId()​

getId(): string

Defined in: packages/orchestration/src/crew/crew-session.ts:245

Get the unique ID of this processable instance

Each processable has a unique ID that persists for its lifetime. IDs are useful for:

  • Filtering specific instances in subscriptions
  • Tracking performance metrics per instance
  • Debugging nested compositions

Returns​

string

Unique ID string

Example​

const bot1 = Baleybot.create({ name: 'analyzer' });
const bot2 = Baleybot.create({ name: 'analyzer' });

bot1.getId();  // 'baleybot-1-a3f891'
bot2.getId();  // 'baleybot-2-b4c2d3'

// Same semantic name, different IDs
bot1.getName();  // 'analyzer'
bot2.getName();  // 'analyzer'

Implementation of​

Processable.getId

getName()​

getName(): string

Defined in: packages/orchestration/src/crew/crew-session.ts:253

What this processable is called. A label, not a key.

Stable across runs, which getId() is not — an id embeds a timestamp and changes on every restart. That is what makes a name safe to print in a log, show on a node, or hand to a merge function.

Not unique. Two bots may both be called 'analyzer', and every unnamed loop() reports 'loop'. Anything that must address exactly one instance uses getId(), or an address — never this. An earlier version of this contract promised uniqueness and had unnamed primitives return their instance id to deliver it; that put generated ids into ParallelMergeInput.botNames and graph node labels, where a name was wanted, and made both change on every restart.

Never returns an empty string: where no name was configured, the primitive reports its kind.

Returns​

string

Implementation of​

Processable.getName

onUpdate()​

onUpdate(callback): () => void

Defined in: packages/orchestration/src/crew/crew-session.ts:146

Register a callback for proactive coordinator updates. Returns an unsubscribe function.

Parameters​

callback​

(message) => void

Returns​

() => void

pause()​

pause(): void

Defined in: packages/orchestration/src/crew/crew-session.ts:157

Pause the crew — stops the scheduler so no new tickets are dispatched. Active runs continue to completion. Call resume() to restart scheduling.

Returns​

void

process()​

process(input, options?): Promise<string>

Defined in: packages/orchestration/src/crew/crew-session.ts:241

Process input and return structured output

You can use BOTH streaming events AND final result together:

  • Streaming events: Provide options.onToken to receive real-time events during execution
  • Final result: Returned when Promise resolves (complete parsed response)

Both work simultaneously - streaming events fire in real-time, then final result is returned.

Parameters​

input​

string

Input data (typed for safety)

options?​

ProcessOptions

Optional callbacks for streaming, completion, errors

Returns​

Promise<string>

Promise resolving to structured output (inferred from TOutput if it's a Zod schema)

Example​

Using both streaming and final result

// Stream events in real-time AND get final result
const result = await bot.process(input, {
  onToken: {
    onTextDelta(botName, event) {
      process.stdout.write(event.content);  // Real-time streaming
    },
    onToolExecutionStart(botName, event) {
      console.log(`Tool starting: ${event.toolName}`);  // Real-time tool status
    }
  }
});

// Also get final result when Promise resolves
console.log('Complete response:', result);  // Final parsed response

Implementation of​

Processable.process

resume()​

resume(): void

Defined in: packages/orchestration/src/crew/crew-session.ts:168

Resume a paused crew — restarts the scheduler which picks up queued tickets and dispatches them.

Returns​

void

send()​

send(message, options?): Promise<string>

Defined in: packages/orchestration/src/crew/crew-session.ts:102

Send a message to the coordinator. Worker completion/failure notifications are automatically injected as context.

Parameters​

message​

string

options?​

ProcessOptions

Returns​

Promise<string>

stop()​

stop(): Promise<CrewResult>

Defined in: packages/orchestration/src/crew/crew-session.ts:184

Stop the session gracefully. Cancels active runs, asks coordinator for a final summary, and returns results.

Returns​

Promise<CrewResult>

subscribeToAll()​

subscribeToAll(_options?): object

Defined in: packages/orchestration/src/crew/crew-session.ts:300

Subscribe to categorized events from this processable and all nested processables

Provides fine-grained control over event handling with categorized callbacks:

  • onStreamEvent: Real-time content streaming (text_delta, structured_output_delta, tool_call_arguments_delta)
  • onProgressUpdate: Status updates (tool_execution_start, tool_call_stream_start, tool_call_stream_complete, reasoning)
  • onComplete: Emitted when each bot finishes processing with final output
  • onError: Error events (error, tool_validation_error)

Callbacks now receive both bot ID and name for precise identification:

  • botId: Unique instance identifier (e.g., 'baleybot-1-a3f891')
  • botName: Semantic name (e.g., 'analyzer', 'router', 'processor')

Parameters​

_options?​
onComplete?​

(botId, botName, output) => void

onError?​

(botId, botName, event) => void

onStreamEvent?​

(botId, botName, event) => void

Returns​

object

Subscription object with unsubscribe method

unsubscribe​

unsubscribe: () => void

Returns​

void

Examples​

Subscribe to all events

const subscription = pipeline.subscribeToAll({
  onStreamEvent(botId, botName, event) {
    // Real-time streaming content
    if (event.type === 'text_delta') {
      console.log(`[${botId}] ${botName}:`, event.content);
    }
  },
  onProgressUpdate(botId, botName, event) {
    // Status updates
    if (event.type === 'tool_execution_start') {
      console.log(`${botName} (${botId}) executing ${event.toolName}`);
    }
  },
  onComplete(botId, botName, output) {
    // Bot finished processing
    console.log(`${botName} completed:`, output);
  },
  onError(botId, botName, event) {
    // Errors
    console.error(`${botName} error:`, event);
  }
});

await pipeline.process('input');
subscription.unsubscribe();

Filter specific bots by name or ID

const subscription = pipeline.subscribeToAll({
  onStreamEvent(botId, botName, event) {
    console.log(`${botName}:`, event);
  },
  omit: ['chat-step', 'processor-42-8a3f91'] // Filter by name or ID
});

Implementation of​

Processable.subscribeToAll