> ## Documentation Index
> Fetch the complete documentation index at: https://docs.modulex.dev/llms.txt
> Use this file to discover all available pages before exploring further.

# Streaming & human-in-the-loop in the SDKs

> Consume ModuleX SSE run streams with listen() async generators, cancel them with AbortSignal or close(), and answer human-in-the-loop prompts with a UserInputResponse in the JavaScript and Python SDKs.

export const MediaEmbed = ({id, type = 'screenshot', caption = '', ext, ratio = '16 / 9'}) => {
  const isVideo = type === 'video' || type === 'app_video';
  const resolvedExt = ext || (isVideo ? 'mp4' : type === 'screenshot' ? 'webp' : 'svg');
  const src = 'https://media.modulex.dev/' + id + '.' + resolvedExt;
  const [status, setStatus] = useState('loading');
  const [isDev, setIsDev] = useState(false);
  const [inView, setInView] = useState(false);
  const boxRef = useRef(null);
  useEffect(() => {
    if (typeof window === 'undefined') return;
    const h = window.location.hostname;
    setIsDev(h === 'localhost' || h === '127.0.0.1' || h.endsWith('.mintlify.app'));
  }, []);
  useEffect(() => {
    if (inView) return;
    if (typeof IntersectionObserver === 'undefined') {
      setInView(true);
      return;
    }
    const el = boxRef.current;
    if (!el) return;
    const io = new IntersectionObserver(entries => {
      if (entries.some(e => e.isIntersecting)) {
        setInView(true);
        io.disconnect();
      }
    }, {
      rootMargin: '300px'
    });
    io.observe(el);
    return () => io.disconnect();
  }, [inView]);
  if (status === 'missing') {
    if (!isDev) return null;
    return <div style={{
      display: 'flex',
      flexDirection: 'column',
      justifyContent: 'center',
      gap: '0.4rem',
      padding: '1rem 1.25rem',
      margin: '1.25rem 0',
      width: '100%',
      aspectRatio: ratio,
      boxSizing: 'border-box',
      border: '1px dashed rgba(128,128,128,0.45)',
      borderRadius: '0.75rem',
      background: 'rgba(128,128,128,0.06)',
      color: 'currentColor',
      fontSize: '0.85rem',
      lineHeight: 1.45
    }}>
        <div style={{
      display: 'flex',
      alignItems: 'center',
      gap: '0.5rem',
      opacity: 0.75
    }}>
          <span aria-hidden="true">🎬</span>
          <code style={{
      fontSize: '0.75rem'
    }}>{id}</code>
          <span style={{
      fontSize: '0.65rem',
      textTransform: 'uppercase',
      letterSpacing: '0.04em',
      padding: '0.1rem 0.4rem',
      borderRadius: '0.4rem',
      background: 'rgba(128,128,128,0.18)'
    }}>
            {type}
          </span>
        </div>
        <div style={{
      opacity: 0.9
    }}>{caption || 'Media not uploaded yet.'}</div>
        <div style={{
      fontSize: '0.7rem',
      opacity: 0.5
    }}>
          Upload to R2 as <code>{id}.{resolvedExt}</code> — preview only, hidden in production.
        </div>
      </div>;
  }
  const mediaStyle = {
    display: status === 'loaded' ? 'block' : 'none',
    width: '100%',
    height: 'auto',
    borderRadius: '0.75rem'
  };
  const media = isVideo ? <video src={inView ? src : undefined} autoPlay loop muted playsInline preload="metadata" onLoadedData={() => setStatus('loaded')} onError={() => setStatus('missing')} style={mediaStyle} /> : <img src={inView ? src : undefined} alt={caption} onLoad={() => setStatus('loaded')} onError={() => setStatus('missing')} style={mediaStyle} />;
  return <figure style={{
    margin: '1.25rem 0'
  }}>
      <div ref={boxRef} style={status === 'loaded' ? {
    width: '100%'
  } : {
    width: '100%',
    aspectRatio: ratio,
    borderRadius: '0.75rem',
    background: 'rgba(128,128,128,0.06)'
  }}>
        {media}
      </div>
      {status === 'loaded' && caption ? <figcaption style={{
    marginTop: '0.5rem',
    textAlign: 'center',
    fontSize: '0.85rem',
    opacity: 0.7
  }}>
          {caption}
        </figcaption> : null}
    </figure>;
};

Every long-running ModuleX operation — a workflow run, an [AI Composer](/concepts/ai-composer) edit session, or an [Assistant](/concepts/assistant) turn — pushes its progress to you over **Server-Sent Events (SSE)**. The [JavaScript SDK](/sdks/javascript) and the [Python SDK](/sdks/python) wrap that transport in `listen()` methods that return **async generators**, so you iterate frames with `for await` (JS) or `async for` (Python). This page covers consuming those streams, cancelling them, and the **human-in-the-loop (HITL)** contract: how a run pauses to ask you a structured question and how you answer it.

For the raw wire format and the per-event payload taxonomy, see [SSE run streaming](/realtime/sse-streaming). For the request/response kinds and resume semantics at the API level, see [Human-in-the-loop (HITL) resume](/realtime/hitl). This page is the SDK view of both.

<Note>
  SSE and Socket.io are two unrelated transports. The `listen()` generators on this page are SSE (run and agent output, served by the ModuleX API). Multi-user canvas collaboration runs over Socket.io and is covered in [Socket.io collaboration events](/realtime/socket-events) — the two share no event names and no envelope.
</Note>

## The streaming model

Each surface follows the same shape: a call mints a `run_id`, and you open a stream for that `run_id`. The stream is one-way (server to client); you resume or cancel through separate calls.

<Steps>
  <Step title="Start the operation">
    Call [`executions.run`](/sdks/javascript), [`composer.chat`](/concepts/ai-composer), or [`assistant.chat`](/concepts/assistant). The response carries a `run_id` (and a `stream_url`).
  </Step>

  <Step title="Open the stream">
    Pass the `run_id` to `listen()`. You get back an async generator that yields parsed frames as they arrive.
  </Step>

  <Step title="Iterate frames">
    Loop over the generator. Switch on the frame's type discriminator. Output frames such as `response_chunk` arrive incrementally; the stream ends on a terminal frame.
  </Step>

  <Step title="Pause for HITL (optional)">
    If a frame of type `user_input_request` arrives, the run is paused and waiting for you. Stop reading and answer it.
  </Step>

  <Step title="Resume or cancel">
    Answer a HITL question with a resume call (which mints a new `run_id` you re-`listen()` on), or stop the run with a cancel call.
  </Step>
</Steps>

<MediaEmbed id="MX-MEDIA-2030" type="image" caption={"Sequence diagram of the SSE listen + HITL resume lifecycle across the SDK and the ModuleX API."} />

## Authentication

Every streaming call authenticates the same way as the rest of the API: an `Authorization: Bearer` header with your `mx_live_*` key, plus an `X-Organization-ID` header naming the organization the run belongs to. The SDKs add both headers for you from the client config; with cURL you set them yourself. SSE GET requests also send `Accept: text/event-stream` and `Cache-Control: no-cache`. See [Authentication](/api-reference/authentication) and [Org context & X-Organization-ID](/security/org-context).

All three run-listen endpoints require the `owner` or `admin` organization role (`organization_admin_required`); the `member` role is retired. See [Roles & permissions](/security/roles-permissions).

## Opening a stream with `listen()`

`listen()` returns an async generator. The first network call happens when you begin iterating, not when you call the method. In JavaScript you consume it with `for await…of`; in Python `listen()` returns an `EventSourceStream` that is both an async iterator and an async context manager — prefer `async with` so the connection always closes.

Workflow runs use [`executions.listen`](/sdks/javascript), which maps to `GET /workflows/listen/{run_id}`.

<CodeGroup>
  ```bash cURL theme={null}
  curl -N https://api.modulex.dev/workflows/listen/run_abc \
    -H "Authorization: Bearer mx_live_xxx" \
    -H "X-Organization-ID: 6f1c2a4e-0b2d-4a8e-9c3f-1d2e3f4a5b6c" \
    -H "Accept: text/event-stream"
  ```

  ```python Python theme={null}
  import asyncio
  from modulex import Modulex

  async def main():
      async with Modulex(api_key="mx_live_xxx", organization_id="6f1c2a4e-0b2d-4a8e-9c3f-1d2e3f4a5b6c") as client:
          async with client.executions.listen("run_abc") as stream:
              async for event in stream:
                  print(event.event, event.data)
                  if event.is_terminal:
                      break

  asyncio.run(main())
  ```

  ```javascript JavaScript theme={null}
  import { Modulex } from "modulex-js";

  const client = new Modulex({
    apiKey: "mx_live_xxx",
    organizationId: "6f1c2a4e-0b2d-4a8e-9c3f-1d2e3f4a5b6c",
  });

  for await (const event of client.executions.listen("run_abc")) {
    if (event.type === "response_chunk") process.stdout.write(event.content ?? "");
    if (event.type === "done" || event.type === "error") break;
  }
  ```
</CodeGroup>

### The five listen surfaces

Every SDK stream is built on the same SSE primitive. The methods differ only in route and payload union.

| Operation      | JavaScript                               | Python                                      | Endpoint                                                |
| -------------- | ---------------------------------------- | ------------------------------------------- | ------------------------------------------------------- |
| Workflow run   | `executions.listen(runId)`               | `executions.listen(run_id)`                 | `GET /workflows/listen/{run_id}`                        |
| Composer turn  | `composer.listen(composerChatId, runId)` | `composer.listen(composer_chat_id, run_id)` | `GET /composer/chat/{composer_chat_id}/listen/{run_id}` |
| Assistant turn | `assistant.listen(chatId, runId)`        | `assistant.listen(chat_id, run_id)`         | `GET /assistant/chat/{chat_id}/listen/{run_id}`         |
| Canvas changes | `workflows.listenChanges(workflowId)`    | `workflows.listen_changes(workflow_id)`     | `GET /workflows/{workflow_id}/changes`                  |
| Chat list      | `chats.stream()`                         | `chats.stream()`                            | `GET /chats/stream`                                     |

There is also one POST-based SSE stream — `credentials.bulkModulexKeys()` (JS) / `credentials.bulk_modulex_keys_stream()` (Python), mapping to `POST /credentials/bulk-modulex-keys/stream` — used for bulk managed-key provisioning.

<ParamField path="runId / run_id" type="string" required>
  The per-execution run identifier returned by `executions.run`, `composer.chat`, `assistant.chat`, or a resume call. This is **not** the run record's `id` (returned by list/get) and **not** the chat id. See the three run-id identities in [Workflows & runs](/concepts/workflows-and-runs).
</ParamField>

<ParamField path="composerChatId / chatId" type="string" required>
  For `composer.listen` and `assistant.listen` only: the chat the run belongs to. The chat id is stable across a conversation; the `run_id` changes per turn and per resume.
</ParamField>

<ParamField path="options.signal (JS) / organization_id" type="AbortSignal | string">
  JavaScript accepts a `RequestOptions` object whose `signal` is an `AbortSignal` for cancellation (see [Cancelling a stream](#cancelling-a-stream)). Both SDKs accept a per-call `organizationId` / `organization_id` that overrides the client default for that stream.
</ParamField>

### What a frame looks like

Run, Composer, and Assistant streams are **data-only**: the server emits `data: {json}\n\n` with no SSE `event:` line, and the real discriminator is the `type` field inside the JSON. (`GET /chats/stream` is the one exception — it uses named `event:` lines.) The SDKs normalize this for you so you always switch on a single field.

<Tabs>
  <Tab title="JavaScript — yielded payload">
    Each `listen()` yields the **payload object** (`frame.data`), cast to a typed union, so you discriminate on `evt.type`:

    ```ts theme={null}
    for await (const evt of client.executions.listen(runId)) {
      switch (evt.type) {
        case "metadata":     /* evt.data: { run_id, thread_id, workflow_name, ... } */ break;
        case "node_started": /* flat: evt.node, evt.name, evt.timestamp */ break;
        case "node_update":  /* flat: evt.node, evt.output */ break;
        case "response_chunk": process.stdout.write(evt.content ?? ""); break;
        case "done":
        case "error": return;
      }
    }
    ```

    Response payloads stay **snake\_case** — only request bodies are converted to snake\_case on the wire, never responses. Read `evt.run_id`, not `evt.runId`.
  </Tab>

  <Tab title="Python — SSEEvent">
    Each `listen()` yields an `SSEEvent` with `.event` (the logical type, normalized), `.data` (parsed JSON dict), `.id`, `.retry`, and `.raw_event`:

    ```python theme={null}
    async for event in stream:
        if event.event == "response_chunk":
            print(event.data.get("data", {}).get("chunk", ""), end="")
        elif event.event == "node_update":
            print(event.data["node"], event.data.get("output"))
        elif event.is_terminal:           # done / error / cancelled / interrupted
            break
    ```

    `SSEEvent.event` always carries the logical name whether the frame used `data["type"]` (run/agent streams) or an SSE `event:` line (`/chats/stream`).
  </Tab>
</Tabs>

<Warning>
  **Payload shape is flat for most events but wrapped for some, and run-event fields are not guaranteed.** Within one run stream, `metadata`, `interrupt`, `resumed`, `done`, `cancelled`, and `user_input_request` are **wrapped** (`{ type, data: {...} }`), while `node_started`, `node_update`, `node_error`, and `error` are **flat** (`{ type, ...fields }`). The Python `WorkflowRunEvent` model is intentionally tolerant (all fields optional, extra allowed) — consume `event.data` defensively and do not assume a field is present. The published wire shape, not the typed models, is authoritative; see [SSE run streaming](/realtime/sse-streaming).
</Warning>

### Terminal and heartbeat frames

The stream ends on a **terminal** frame. The exact set differs by layer, which matters when you write a break condition:

* A workflow run stream (`GET /workflows/listen/{run_id}`) is closed server-side only on `done` or `error`.
* The Python SDK's `SSEEvent.is_terminal` is broader — `True` for `done`, `error`, `cancelled`, and `interrupted` — so it may stop iterating on `cancelled` even where the backend would keep the socket open.
* The JavaScript SDK does not expose an `is_terminal` helper; break on `done` / `error` yourself (and on `cancelled` if you cancelled the run).

`heartbeat` frames (`{"type":"heartbeat"}`) are injected about every 15 seconds of idle to keep the connection warm. The Python SDK **filters them out** unless you construct the stream with `include_heartbeats=True`. In JavaScript a `heartbeat` data frame is yielded as a normal event — ignore it in your switch. `/chats/stream` instead emits an SSE comment keepalive every 30 seconds, which both SDKs skip.

<Warning>
  `interrupt` (a workflow pause) and `user_input_request` (a Composer/Assistant HITL pause) are **not** terminal. After either, the stream goes quiet but is not closed and no `done` arrives — the run is paused waiting for you. Do not treat silence as completion; handle the pause frame explicitly (see [Human-in-the-loop](#human-in-the-loop)).
</Warning>

## Cancelling a stream

You can stop reading at any time, and you can also cancel the underlying connection. The mechanism differs by SDK.

In **JavaScript**, pass an `AbortSignal` in the request options. Aborting it stops the `fetch`, breaks the parser's read loop, and releases the reader cleanly. Note that SSE streams bypass the client's `timeout` and `maxRetries` — only your `signal` cancels them.

In **Python**, use `async with` (the context manager closes the connection on exit) or call `close()` to break the loop. The SDK disables the read timeout on streams (so a long, idle HITL pause is not killed by a read timeout); the connect/write timeout still applies.

Stopping the stream does **not** stop the run. To actually terminate a run server-side, call the cancel method, which posts to the cancel endpoint and emits a `cancelled` frame.

<CodeGroup>
  ```bash cURL theme={null}
  # Stop the server-side run (separate from closing the stream connection)
  curl -X POST https://api.modulex.dev/workflows/cancel/run_abc \
    -H "Authorization: Bearer mx_live_xxx" \
    -H "X-Organization-ID: 6f1c2a4e-0b2d-4a8e-9c3f-1d2e3f4a5b6c" \
    -H "Content-Type: application/json" \
    -d '{"reason": "user_requested"}'
  ```

  ```python Python theme={null}
  import asyncio
  from modulex import Modulex

  async def main():
      async with Modulex(api_key="mx_live_xxx", organization_id="org_uuid") as client:
          stream = client.executions.listen("run_abc")
          async with stream:
              async for event in stream:
                  if event.event == "node_update":
                      await stream.close()          # stop reading the stream
                      break
          # Terminate the run itself, server-side:
          await client.executions.cancel("run_abc", reason="user_requested")

  asyncio.run(main())
  ```

  ```javascript JavaScript theme={null}
  const controller = new AbortController();
  const stream = client.executions.listen("run_abc", { signal: controller.signal });

  // Stop listening after 5 seconds
  setTimeout(() => controller.abort(), 5_000);

  try {
    for await (const evt of stream) {
      if (evt.type === "done" || evt.type === "error") break;
    }
  } catch (err) {
    // Aborting surfaces as the fetch's abort error — handle or ignore as needed
  }

  // Terminate the run itself, server-side:
  await client.executions.cancel("run_abc", { reason: "user_requested" });
  ```
</CodeGroup>

Cancel methods exist for each surface: `executions.cancel(runId)` (`POST /workflows/cancel/{run_id}`), `composer.cancel(composerChatId)` (`POST /composer/chat/{composer_chat_id}/cancel`), and `assistant.cancel(chatId)` (`POST /assistant/chat/{chat_id}/cancel`). Assistant cancel also clears any pending HITL question. Each raises a not-found error if the run or chat does not exist or is not owned by your organization.

## Human-in-the-loop

A Composer or Assistant run can pause to ask you a structured question — pick an option, confirm a yes/no, supply free text, or connect a credential. The pause arrives as a `user_input_request` frame; you answer with a resume call carrying a `UserInputResponse`. This is the SDK view; the wire contract is in [Human-in-the-loop (HITL) resume](/realtime/hitl).

<MediaEmbed id="MX-MEDIA-2031" type="image" caption={"The five HITL request kinds mapped to their seven response kinds."} />

### The `user_input_request` frame

This frame is **doubly wrapped**: the SSE frame's `data` is `{ "type": "user_input_request", "data": <UserInputRequest> }`, so the actual question lives one level deeper at `data.data`. A naive read of `event.data["kind"]` returns nothing — the kind is at `event.data["data"]["kind"]`.

* In **JavaScript**, the yielded `user_input_request` event is typed `{ type: "user_input_request"; data: UserInputRequest }`, so the question is `evt.data`.
* In **Python**, use the helper `user_input_request_from_event(event.data)`, which extracts and parses the nested question into the right subtype (and returns `None` if the frame is not a HITL question).

<ResponseField name="UserInputRequest" type="object">
  The HITL question, discriminated on `kind`. Base fields (all kinds): `request_id` (string), `message` (markdown), `required` (boolean, default `true`), `allow_free_text` (boolean, default `false`), `context` (object, optional), `timeout_hint_seconds` (integer, optional).

  <Expandable title="request kinds (discriminate on kind)">
    <ResponseField name="single_choice" type="object">
      Pick exactly one option. Extra field: `options` (array of `ChoiceOption`, 1–10).
    </ResponseField>

    <ResponseField name="multi_choice" type="object">
      Pick several. Extra fields: `options` (array of `ChoiceOption`, 1–20), `min_selections` (integer), `max_selections` (integer, optional).
    </ResponseField>

    <ResponseField name="yes_no" type="object">
      A boolean confirmation. Extra fields: `yes_label` (default `Yes`), `no_label` (default `No`).
    </ResponseField>

    <ResponseField name="free_text" type="object">
      Open input. Extra fields: `placeholder` (optional), `multiline` (boolean), `min_length` (integer), `max_length` (integer, optional).
    </ResponseField>

    <ResponseField name="credential_request" type="object">
      Connect a credential for an integration. Extra fields: `integration_name`, `integration_display_name`, `integration_logo` (optional), `auth_options` (array of `CredentialAuthOption`), `pending_node_name` (optional). Each `CredentialAuthOption.auth_type` is one of `oauth2`, `api_key`, `bearer_token`, `modulex_key`, `custom`.
    </ResponseField>
  </Expandable>

  `ChoiceOption` is `{ value, label, description?, icon?, badge? }`.
</ResponseField>

### Answering with a `UserInputResponse`

You answer with a resume call. The response is also discriminated on `kind` — there are **seven** response kinds for the five request kinds, because `credential_request` is answered by either `credential_added` or `credential_failed`, and `skipped` can answer any non-required question.

<ResponseField name="UserInputResponse" type="object">
  The answer, discriminated on `kind`.

  <Expandable title="response kinds (discriminate on kind)">
    <ResponseField name="single_choice" type="object">
      Fields: `selected_value` (string, optional), `free_text` (string, optional).
    </ResponseField>

    <ResponseField name="multi_choice" type="object">
      Field: `selected_values` (array of strings).
    </ResponseField>

    <ResponseField name="yes_no" type="object">
      Field: `answer` (boolean).
    </ResponseField>

    <ResponseField name="free_text" type="object">
      Field: `text` (string).
    </ResponseField>

    <ResponseField name="credential_added" type="object">
      Fields: `credential_id`, `integration_name`, `auth_type`. The server runs a preflight validation on the credential before the agent commits to it.
    </ResponseField>

    <ResponseField name="credential_failed" type="object">
      Fields: `integration_name`, `auth_type`, `error_code`, `error_message`, `retryable` (boolean, default `true`), `provider_details` (optional). `error_code` is one of `oauth_denied`, `oauth_provider_error`, `invalid_credentials`, `network_error`, `popup_closed`, `timeout`, `unknown`.
    </ResponseField>

    <ResponseField name="skipped" type="object">
      Field: `reason` (string, optional).
    </ResponseField>
  </Expandable>
</ResponseField>

### Resume mints a new `run_id`

The most important HITL fact: **a resume call returns a NEW `run_id`** and a fresh `stream_url`. The original stream does **not** carry the resumed events — you must call `listen()` again on the new `run_id`. The chat id stays the same; only the run changes.

<ParamField path="composer_chat_id / chatId" type="string" required>
  Path argument — the chat whose paused question you are answering.
</ParamField>

<ParamField path="request_id" type="string" required>
  The `request_id` from the `user_input_request` you are answering. Resuming with a stale or already-answered `request_id` returns `410 Gone`.
</ParamField>

<ParamField path="response" type="UserInputResponse" required>
  The answer object, discriminated on `kind` (above). In Python you may pass a typed model (e.g. `YesNoResponse(answer=True)`) or a plain dict.
</ParamField>

<ParamField path="llm" type="ComposerLLMConfig" required>
  The model config to continue with: `{ integration_name, provider_id, model_id, temperature?, credential_id? }`. **Required in practice** — the endpoint returns `400` if it is missing, because the executor rebuilds the chat model on resume. The two SDKs reflect this differently: `assistant.resume` makes `llm` a required argument with no default, while `composer.resume` types it as optional but the server still rejects a missing value. Always send it.
</ParamField>

<CodeGroup>
  ```bash cURL theme={null}
  # 1. Answer the paused question (returns a NEW run_id + stream_url)
  curl -X POST https://api.modulex.dev/composer/chat/cmp_123/resume \
    -H "Authorization: Bearer mx_live_xxx" \
    -H "X-Organization-ID: org_uuid" \
    -H "Content-Type: application/json" \
    -d '{
      "request_id": "req_42",
      "response": { "kind": "yes_no", "answer": true },
      "llm": { "integration_name": "openai", "provider_id": "openai", "model_id": "gpt-4o" }
    }'

  # 2. Re-listen on the NEW run_id from the response
  curl -N https://api.modulex.dev/composer/chat/cmp_123/listen/<new_run_id> \
    -H "Authorization: Bearer mx_live_xxx" \
    -H "X-Organization-ID: org_uuid" \
    -H "Accept: text/event-stream"
  ```

  ```python Python theme={null}
  from modulex.types.realtime import user_input_request_from_event, YesNoResponse, ComposerLLMConfig

  llm = ComposerLLMConfig(integration_name="openai", provider_id="openai", model_id="gpt-4o")
  resp = await client.composer.chat("Build me a Slack alert workflow", llm=llm)
  run_id = resp.run_id

  while True:
      paused = None
      async with client.composer.listen(resp.composer_chat_id, run_id) as stream:
          async for event in stream:
              if event.event == "user_input_request":
                  paused = user_input_request_from_event(event.data)
                  break                       # stop reading; go answer it
              if event.is_terminal:
                  break                       # run finished
      if paused is None:
          break

      new = await client.composer.resume(
          resp.composer_chat_id,
          request_id=paused.request_id,
          response=YesNoResponse(answer=True),
          llm=llm,
      )
      run_id = new.run_id                      # NEW run_id — loop re-listens on it
  ```

  ```javascript JavaScript theme={null}
  const llm = { integration_name: "openai", provider_id: "openai", model_id: "gpt-4o" };
  const start = await client.composer.chat({ message: "Build me a Slack alert workflow", llm });
  let runId = start.run_id;

  for (;;) {
    let paused = null;
    for await (const evt of client.composer.listen(start.composer_chat_id, runId)) {
      if (evt.type === "user_input_request") {
        paused = evt.data;                     // UserInputRequest; discriminate on paused.kind
        break;
      }
      if (evt.type === "done" || evt.type === "error") break;
    }
    if (!paused) break;

    const resumed = await client.composer.resume(start.composer_chat_id, {
      requestId: paused.request_id,
      response: { kind: "yes_no", answer: true },
      llm,
    });
    runId = resumed.run_id;                    // NEW run_id — loop re-listens on it
  }
  ```
</CodeGroup>

<Note>
  Two distinct "resume" flows exist — do not conflate them. **HITL chat resume** (`composer.resume` / `assistant.resume`) answers a `user_input_request` and mints a **new** `run_id`. **Workflow thread resume** (`executions.resume`, `POST /workflows/resume/{thread_id}`) resumes a workflow `interrupt` node with a `resume_value` and **reuses** the same `run_id`. The first is for chat agents; the second is for the [interrupt node](/workflow-builder/nodes/interrupt) in a workflow.
</Note>

## Error handling

A non-2xx response on connect throws synchronously **before the first frame** — it never arrives as a yielded event. The SDKs map status codes to typed exceptions.

| Status      | JavaScript class                    | Python class                             | When                                                                                                      |
| ----------- | ----------------------------------- | ---------------------------------------- | --------------------------------------------------------------------------------------------------------- |
| 401         | `AuthenticationError`               | `AuthenticationError`                    | Missing or invalid API key                                                                                |
| 402         | `ModulexError` (no dedicated class) | `PaymentRequiredError` / `BillingError`  | Billing gate: credits exhausted or wallet overage denied                                                  |
| 403         | `PermissionError`                   | `PermissionError` / `QuotaExceededError` | Not `owner`/`admin`, not the question's originator on resume, or quota denied                             |
| 404         | `NotFoundError`                     | `NotFoundError`                          | Run or chat does not exist **or** is not owned by your org — identical 404 either way (no existence leak) |
| 409         | `ConflictError`                     | `ConflictError`                          | A `chat()` while a HITL question is pending or a run is already in progress                               |
| 410         | `ModulexError` (no dedicated class) | `ModulexError` (no dedicated class)      | Resume: `request_id` not pending or already consumed                                                      |
| 429         | `RateLimitError`                    | `RateLimitError` / `BillingError`        | Rate limit or the `rate`-layer billing denial                                                             |
| (body null) | `StreamError`                       | `StreamError`                            | The SSE response body could not be read; or, in Python, any mid-stream failure                            |

Discriminate by `instanceof` (JS) / `isinstance` (Python), not by class-name string — for example `403` maps to `PermissionError`, not "ForbiddenError".

A backend failure **mid-stream** does not throw. It arrives as a normal yielded frame of `type === "error"` (JS) / `event == "error"` (Python) — your loop must handle it.

<Warning>
  Composer, Assistant, and workflow runs sit behind the **live billing admission gate**. A `chat()`, `run()`, or `resume()` that exceeds your plan can return a flat `DenialEnvelope` — `{code, layer, key, current, limit, reason}` — as **402**, **403**, or **429** (this is distinct from the `{detail}` shape returned by plain CRUD routes). Surface and branch on these in your error handler. See [Usage gating & limits](/billing/usage-gating) and [Errors & status codes](/api-reference/errors).
</Warning>

For the full SDK error tree, retry policy, and idempotency behavior, see [Errors & retries](/sdks/errors-retries).

## Reconnecting

Neither SDK auto-reconnects a dropped stream. On the server side, a run's recent events are replayed from a one-hour history buffer when you reconnect by calling `listen()` again with the same `run_id`, so a brief disconnect during an active run does not lose frames. The Python `EventSourceStream` records `last_event_id`, but the SDKs do not replay it via a `Last-Event-ID` header — reconnection relies on the server-side history replay, not on SSE `id:`. See [SSE run streaming](/realtime/sse-streaming) for the replay window.

## Related

<CardGroup cols={2}>
  <Card title="SSE run streaming" icon="radio" href="/realtime/sse-streaming">
    The raw frame format, the full event taxonomy, and the heartbeat and replay model.
  </Card>

  <Card title="Human-in-the-loop (HITL) resume" icon="hand" href="/realtime/hitl">
    Request and response kinds, the resume endpoints, and the new-run\_id contract.
  </Card>

  <Card title="JavaScript SDK" icon="square-js" href="/sdks/javascript">
    Install and configure modulex-js, including headers and config.
  </Card>

  <Card title="Python SDK" icon="python" href="/sdks/python">
    Install and configure the modulex-python async client.
  </Card>
</CardGroup>