context.streams = {
  askAI: [
    // [0] = most recent invocation
    {
      actionId: null, // Tracks which action triggered this stream (for error tracing)
      loading: false,
      payload: null, // Evaluated payload sent with stream
      response: {
        data: [], // Array of deserialized chunk payloads (grows during stream)
        usage: null, // { promptTokens, completionTokens } — set on complete
        finishReason: null, // 'stop' | 'cancelled' | 'error' — set on complete
      },
      error: null,
      responseTime: null, // ms from start to completion
      streamId: 'askAI',
    },
    // [1] = previous invocation, etc.
  ],
};
// Internal runtime properties (not exposed via _stream/_stream_details):
// - abortController: AbortController instance, set during active streaming,
//   deleted on completion. Stored on the stream object for cancel+restart
//   but excluded from operator access.

// Pseudocode for Streams.fetchStream()
// Follows same actionId threading pattern as Requests.fetch()
async fetchStream(streamDef, { actions, actionId, arrayIndices, event }) {
  const streamHistory = this.context.streams[streamDef.streamId];

  // Cancel + restart: if the most recent invocation is still active, abort it first
  if (streamHistory[0]?.loading && streamHistory[0]?.abortController) {
    streamHistory[0].abortController.abort();
  }

  // Evaluate payload operators (same as Requests.callRequest)
  const { output: payload, errors: parserErrors } = this.context._internal.parser.parse({
    actions,
    event,
    arrayIndices,
    input: streamDef.payload,
    location: streamDef.streamId,
  });
  if (parserErrors.length > 0) {
    throw parserErrors[0];
  }

  // Create new stream entry and unshift to history (same pattern as Requests)
  const stream = {
    actionId,
    loading: true,
    payload,
    response: { data: [], usage: null, finishReason: null },
    error: null,
    responseTime: null,
    streamId: streamDef.streamId,
  };
  streamHistory.unshift(stream);

  // Per-run token: prevents stale writes from a previous run's catch/finally
  // blocks after cancel+restart. All state writes check this before proceeding.
  const runId = actionId;
  const isCurrentRun = () => streamHistory[0]?.actionId === runId;

  this.context._internal.update();

  const abortController = new AbortController();
  stream.abortController = abortController;
  const startTime = Date.now();

  // Matches request pattern: engine serializes payload, fetch layer stringifies body
  const serializedPayload = serializer.serialize(payload);
  const fetchResponse = await this.context._internal.lowdefy._internal.callStream({
    actionId,
    pageId: this.context.pageId,
    streamId: streamDef.streamId,
    payload: serializedPayload,
    signal: abortController.signal,
  });
  // callStream internally does:
  //   fetch(url, {
  //     method: 'POST',
  //     headers: { 'Content-Type': 'application/json' },
  //     body: JSON.stringify({ payload }),
  //     signal,
  //   })
  // Returns the raw Response (not parsed) so we can read the stream

  // Guard: validate response before opening stream reader.
  // Pre-SSE failures (401, 403, 500) return JSON, not SSE.
  if (!fetchResponse.ok || !fetchResponse.headers.get('content-type')?.includes('text/event-stream')) {
    const errorBody = await fetchResponse.json().catch(() => ({}));
    throw new Error(errorBody.message ?? `Stream request failed: ${fetchResponse.status}`);
  }

  const reader = fetchResponse.body.getReader();
  const decoder = new TextDecoder();
  const throttleMs = Math.max((streamDef.client?.throttleRender ?? 500), 100);

  let dataBuffer = [];
  let dirty = false;
  let flushTimer = null;
  let lineBuffer = ''; // Accumulates partial lines across reader.read() chunks
  const MAX_LINE_BUFFER = 15 * 1024 * 1024; // 15MB guard against malformed streams

  // Note: on cancel+restart, the previous run's flushTimer may still be pending.
  // It's a local variable in the previous call's scope and can't be cleared from here.
  // The isCurrentRun() guard ensures it returns early without writing stale data.
  const flush = () => {
    if (!isCurrentRun()) return; // Stale run — discard
    if (dirty) {
      stream.response.data.push(...dataBuffer);
      dataBuffer = [];
      dirty = false;
      this.context._internal.update();
    }
    flushTimer = null;
  };

  const scheduleFlush = () => {
    if (!flushTimer) {
      flushTimer = setTimeout(flush, throttleMs);
    }
  };

  // Process a complete SSE line
  const processLine = (line) => {
    if (!line.startsWith('data: ')) return;
    const raw = line.slice(6);
    if (raw === '[DONE]') return;

    const chunk = JSON.parse(raw);
    if (chunk.type === 'chunk') {
      // Deserialize chunk payload (preserves Date, ObjectId, etc.)
      const deserialized = serializer.deserialize(chunk.payload);
      dataBuffer.push(deserialized.data);
      dirty = true;
      scheduleFlush();
    } else if (chunk.type === 'complete') {
      flush();
      if (isCurrentRun()) {
        const meta = serializer.deserialize(chunk.payload);
        stream.response.usage = meta.usage;
        stream.response.finishReason = meta.finishReason;
      }
    } else if (chunk.type === 'error') {
      throw new Error(chunk.message);
    }
  };

  try {
    while (true) {
      const { done, value } = await reader.read();
      if (done) break;

      // Accumulate into line buffer, then split on newlines.
      // SSE events can be split across reader.read() chunks at arbitrary
      // byte boundaries, so we must buffer incomplete lines.
      lineBuffer += decoder.decode(value, { stream: true });
      if (lineBuffer.length > MAX_LINE_BUFFER) {
        throw new Error('Stream line buffer exceeded maximum size.');
      }
      const lines = lineBuffer.split('\n');
      // Last element is either '' (line was complete) or a partial line
      lineBuffer = lines.pop();
      for (const line of lines) {
        processLine(line);
      }
    }
    // Flush any remaining bytes from TextDecoder (multibyte edge case)
    lineBuffer += decoder.decode();
    if (lineBuffer) {
      processLine(lineBuffer);
    }
  } catch (error) {
    if (error.name === 'AbortError') {
      stream.response.finishReason = 'cancelled';
    } else {
      stream.error = error;
      stream.response.finishReason = 'error';
    }
  } finally {
    // Release the reader to free the underlying connection.
    // For AbortError the browser cleans up, but for other errors
    // (JSON parse, SSE error event) the reader may still be locked.
    try { reader.cancel(); } catch (_) { /* already closed */ }
  }

  // Clean up stream-local state regardless of whether this is the current run.
  // On cancel+restart, the new run has already unshifted a new entry, so this
  // stream object is now at streamHistory[1+]. We still need correct state on
  // historical entries (loading: false, finishReason set, no abortController).
  if (flushTimer) clearTimeout(flushTimer);
  if (dirty) {
    stream.response.data.push(...dataBuffer);
  }
  stream.loading = false;
  stream.responseTime = Date.now() - startTime;
  delete stream.abortController;

  // Only trigger framework side-effects (re-render, error propagation) for
  // the current run. Stale runs must not call update() or throw errors that
  // could interfere with the active run.
  if (!isCurrentRun()) return;

  this.context._internal.update();

  // If cancelled or errored, rethrow so blocking Stream action rejects
  // (stops subsequent actions in the chain). Partial results are preserved in state.
  if (stream.response.finishReason === 'cancelled') {
    const cancelError = new Error('Stream cancelled.');
    cancelError.name = 'StreamCancelledError';
    throw cancelError;
  }
  if (stream.error) {
    throw stream.error;
  }
}

import { ConfigError, PluginError, ServiceError } from '@lowdefy/errors/server';
import { serializer } from '@lowdefy/helpers';

async function handler({ context, req, res }) {
  const { logger } = context;
  const { pageId, streamId } = req.query;
  const { payload } = req.body;

  // Steps 1–8: shared with request pipeline.
  // prepareStream is extracted from the existing callRequest.js during implementation —
  // the shared logic (config loading, auth, operator evaluation, schema validation) is
  // factored into a common prepare function, with callRequest and callStream each calling
  // it with their respective config paths (requests/ vs streams/).
  const prepared = await prepareStream(context, { pageId, streamId, payload });
  const { streamConfig, streamResolver, connectionProperties, streamProperties } = prepared;

  // SSE headers
  res.writeHead(200, {
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache',
    Connection: 'keep-alive',
    'X-Accel-Buffering': 'no',
  });

  // Abort on client disconnect
  const abortController = new AbortController();
  req.on('close', () => abortController.abort());

  // Framework owns serialization: plugins call onChunk({ data }) with raw values,
  // the wrapper serializes before writing to the SSE stream.
  // Plugins MAY pre-process provider-specific types (e.g., MongoDB ObjectId → _oid)
  // using their own serialize utilities before calling onChunk. The framework
  // serializer then handles standard type markers (~d for Date, ~e for Error).
  function onChunk({ data }) {
    if (!abortController.signal.aborted) {
      const serialized = serializer.serialize({ data });
      res.write(`data: ${JSON.stringify({ type: 'chunk', payload: serialized })}\n\n`);
    }
  }

  try {
    const metadata = await streamResolver({
      connection: connectionProperties,
      request: streamProperties,
      payload: context.payload,
      onChunk,
      signal: abortController.signal,
    });

    if (!abortController.signal.aborted) {
      const serialized = serializer.serialize(metadata ?? {});
      res.write(`data: ${JSON.stringify({ type: 'complete', payload: serialized })}\n\n`);
    }
  } catch (error) {
    // Follow callRequestResolver.js error wrapping pattern:
    // 1. Attach configKey for location tracing
    if (!error.configKey) {
      error.configKey = streamConfig['~k'];
    }
    // 2. ConfigError passes through as-is
    // 3. Service errors (network, timeout) wrapped in ServiceError
    // 4. Everything else wrapped in PluginError
    let wrappedError;
    if (error instanceof ConfigError) {
      wrappedError = error;
    } else if (ServiceError.isServiceError(error)) {
      wrappedError = new ServiceError({
        error,
        service: streamConfig.connectionId,
        configKey: streamConfig['~k'],
      });
    } else {
      wrappedError = new PluginError({
        error,
        pluginType: 'stream',
        pluginName: streamConfig.type,
        received: streamProperties,
        location: `${streamConfig.connectionId}/${streamConfig.streamId}`,
        configKey: streamConfig['~k'],
      });
    }

    if (!abortController.signal.aborted) {
      res.write(`data: ${JSON.stringify({ type: 'error', message: wrappedError.message })}\n\n`);
    }
    logger.debug({ err: wrappedError }, wrappedError.message);
    // Do NOT re-throw: headers are already sent (200 + SSE), so apiWrapper
    // cannot send an HTTP error response. Errors are communicated via SSE events.
  }

  res.end();
}

export default apiWrapper(handler);

// connection-openai/src/connections/OpenAIChat/OpenAIChat.js
import OpenAIChatCompletion from './OpenAIChatCompletion/OpenAIChatCompletion.js';
import OpenAIChatStream from './OpenAIChatStream/OpenAIChatStream.js';

export default {
  schema: connectionSchema,
  requests: {
    OpenAIChatCompletion, // Non-streaming (existing pattern)
  },
  streams: {
    OpenAIChatStream, // Streaming (new)
  },
};

async function OpenAIChatStream({ request, connection, onChunk, signal }) {
  // Resolvers must propagate abort signal to provider calls where supported.
  // stream_options.include_usage is required to get token usage in streaming mode —
  // OpenAI only sends usage in the final chunk when this option is set.
  const stream = await openai.chat.completions.create(
    {
      model: request.model ?? connection.model,
      messages: request.messages,
      max_tokens: request.maxTokens,
      temperature: request.temperature,
      stream: true,
      stream_options: { include_usage: true },
    },
    { signal }
  );

  // Collect metadata from chunks during iteration.
  // OpenAI streaming sends usage in the final chunk and finishReason per-choice.
  let usage = null;
  let finishReason = 'stop';

  for await (const chunk of stream) {
    if (signal?.aborted) break;

    const choice = chunk.choices[0];
    if (choice?.delta?.content) {
      onChunk({ data: choice.delta.content });
    }
    if (choice?.finish_reason) {
      finishReason = choice.finish_reason;
    }
    if (chunk.usage) {
      usage = {
        promptTokens: chunk.usage.prompt_tokens,
        completionTokens: chunk.usage.completion_tokens,
      };
    }
  }

  return { usage, finishReason };
}

OpenAIChatStream.schema = streamRequestSchema;
OpenAIChatStream.meta = {
  checkRead: true,
  checkWrite: false,
};

export default OpenAIChatStream;

return {
  usage: { promptTokens: number, completionTokens: number },
  finishReason: 'stop' | 'length' | 'tool_calls' | string,
  // Any additional metadata the plugin wants to expose
};

import * as connections from './connections.js';

export default {
  connections: Object.keys(connections),
  requests: Object.keys(connections)
    .map((c) => Object.keys(connections[c].requests ?? {}))
    .flat(),
  streams: Object.keys(connections)
    .map((c) => Object.keys(connections[c].streams ?? {}))
    .flat(),
};

import { type } from '@lowdefy/helpers';
import { ConfigError } from '@lowdefy/errors/build';

function buildStream(stream, pageContext) {
  const { auth, checkDuplicateStreamId, context, pageId, typeCounters } = pageContext;
  const configKey = stream['~k'];

  if (type.isUndefined(stream.id)) {
    throw new ConfigError({
      message: `Stream id missing at page "${pageId}".`,
      configKey,
      context,
    });
  }
  if (!type.isString(stream.id)) {
    throw new ConfigError({
      message: `Stream id is not a string at page "${pageId}".`,
      received: stream.id,
      configKey,
      context,
    });
  }
  checkDuplicateStreamId(stream);

  // Stream and request IDs must be unique across both types on the same page
  if (pageContext.requestIds.has(stream.id)) {
    throw new ConfigError({
      message: `Stream "${stream.id}" at page "${pageId}" has the same id as a request.`,
      configKey,
      context,
    });
  }

  // connectionId is required for streams
  if (type.isNone(stream.connectionId)) {
    throw new ConfigError({
      message: `Stream "${stream.id}" at page "${pageId}" connectionId is missing.`,
      configKey,
      context,
    });
  }
  if (!type.isString(stream.connectionId)) {
    throw new ConfigError({
      message: `Stream "${stream.id}" at page "${pageId}" connectionId is not a string.`,
      received: stream.connectionId,
      configKey,
      context,
    });
  }
  if (!context.connectionIds.has(stream.connectionId)) {
    throw new ConfigError({
      message: `Stream "${stream.id}" at page "${pageId}" references non-existent connection "${stream.connectionId}".`,
      configKey,
      context,
      checkSlug: 'connection-refs',
    });
  }

  // Set defaults (use type.isNone to catch both null and undefined)
  if (type.isNone(stream.payload)) stream.payload = {};
  if (type.isNone(stream.client)) stream.client = {};
  if (type.isNone(stream.client.throttleRender)) stream.client.throttleRender = 500;

  stream.auth = auth;
  stream.streamId = stream.id;
  stream.pageId = pageId;
  stream.id = `stream:${pageId}:${stream.id}`;
  pageContext.streams.push(stream);
}

// connection-mongodb/src/connections/MongoDBCollection/MongoDBCollectionAggregateStream.js

import getCollection from '../getCollection.js';
import { serialize, deserialize } from '../serialize.js';

async function MongoDBCollectionAggregateStream({ request, connection, onChunk, signal }) {
  const deserializedRequest = deserialize(request);
  const { pipeline, options } = deserializedRequest;

  const { collection, client } = await getCollection({ connection });

  let documentCount = 0;

  try {
    const cursor = collection.aggregate(pipeline, options);

    for await (const document of cursor) {
      if (signal?.aborted) break;

      // Pre-process MongoDB types (ObjectId → _oid) using plugin's serialize utility.
      // Framework's onChunk wrapper then handles standard types (~d for Date, ~e for Error).
      // Same two-layer pattern as request resolvers.
      onChunk({ data: serialize(document) });
      documentCount++;
    }
  } finally {
    await client.close();
  }

  return {
    finishReason: signal?.aborted ? 'cancelled' : 'stop',
    documentCount,
  };
}

MongoDBCollectionAggregateStream.schema = {
  $schema: 'http://json-schema.org/draft-07/schema#',
  type: 'object',
  required: ['pipeline'],
  properties: {
    pipeline: {
      type: 'array',
      description: 'MongoDB aggregation pipeline stages.',
    },
    options: {
      type: 'object',
      description: 'Aggregation options (e.g., allowDiskUse).',
    },
  },
};

MongoDBCollectionAggregateStream.meta = {
  checkRead: true,
  checkWrite: false,
};

export default MongoDBCollectionAggregateStream;

// connection-mongodb/src/connections/MongoDBCollection/MongoDBCollection.js
import MongoDBFind from './MongoDBFind/MongoDBFind.js';
import MongoDBInsertOne from './MongoDBInsertOne/MongoDBInsertOne.js';
import MongoDBAggregate from './MongoDBAggregate/MongoDBAggregate.js';
import MongoDBCollectionAggregateStream from './MongoDBCollectionAggregateStream/MongoDBCollectionAggregateStream.js';
// ... other imports

export default {
  schema: connectionSchema,
  requests: {
    MongoDBFind,
    MongoDBInsertOne,
    MongoDBAggregate, // Non-streaming: returns complete result array
    // ... other request resolvers
  },
  streams: {
    MongoDBCollectionAggregateStream, // Streaming: yields documents one by one
  },
};

// Accumulate partial lines across reader.read() chunks
lineBuffer += decoder.decode(value, { stream: true });
const lines = lineBuffer.split('\n');
lineBuffer = lines.pop(); // Keep incomplete trailing line

for (const line of lines) {
  if (!line.startsWith('data: ')) continue;
  const chunk = JSON.parse(line.slice(6));
  switch (chunk.type) {
    case 'chunk': {
      const deserialized = serializer.deserialize(chunk.payload);
      dataBuffer.push(deserialized.data);
      break;
    }
    case 'complete': {
      const meta = serializer.deserialize(chunk.payload);
      /* set usage, finishReason from meta */
      break;
    }
    case 'error':
      /* throw */ break;
  }
}

import { streamText } from 'ai';
import { openai } from '@ai-sdk/openai';

async function OpenAIAdvancedStream({ request, connection, onChunk, signal }) {
  const result = streamText({
    model: openai(connection.model),
    messages: request.messages,
    maxTokens: request.maxTokens,
    abortSignal: signal,
  });

  // Consume the AI SDK stream and map to Lowdefy onChunk calls
  for await (const event of result.fullStream) {
    if (signal?.aborted) break;

    switch (event.type) {
      case 'text-delta':
        // Text delta → pushed onto response.data as structured object
        onChunk({ data: { type: 'text', content: event.textDelta } });
        break;

      case 'reasoning':
        // Reasoning → pushed onto response.data as structured object
        onChunk({ data: { type: 'reasoning', text: event.textDelta } });
        break;

      case 'tool-call':
        // Tool calls → pushed onto response.data
        onChunk({
          data: {
            type: 'tool_call',
            toolName: event.toolName,
            args: event.args,
            toolCallId: event.toolCallId,
          },
        });
        break;

      case 'tool-result':
        // Tool results → pushed onto response.data
        onChunk({
          data: {
            type: 'tool_result',
            toolCallId: event.toolCallId,
            result: event.result,
          },
        });
        break;

      case 'source':
        // Citations → pushed onto response.data
        onChunk({
          data: {
            type: 'citation',
            url: event.source.url,
            title: event.source.title,
          },
        });
        break;
    }
  }

  const usage = await result.usage;
  const finishReason = await result.finishReason;

  return {
    usage: {
      promptTokens: usage.promptTokens,
      completionTokens: usage.completionTokens,
    },
    finishReason,
  };
}