Function frameFromEnvelope

node/src/streaming.ts:59–109

(env: Envelope)

      const env = result.value;
      if (env.kind === Envelope_Kind.EVENT && env.payload.case === "askEvent") {
        let payload: unknown = env.payload.value.payloadJson;
        try {
          payload = JSON.parse(env.payload.value.payloadJson);
        } catch {
          // Preserve a non-JSON payload verbatim.
        }
        yield {
          type: "event",
          eventType: env.payload.value.eventType,
          payload,
          machineId: env.payload.value.machineId,
        };
      } else if (env.kind === Envelope_Kind.EVENT && env.payload.case === "askStreamError") {
        yield {
          type: "error",
          code: STREAM_ERROR_NAMES[env.payload.value.code] ?? "internal",
          message: env.payload.value.message,
        };
      } else if (env.kind === Envelope_Kind.DONE && env.payload.case === "askDone") {
        const value = env.payload.value;
        yield { type: "done", ...value };
      }
    }
  }

  async collect(): Promise<string> {
    let text = "";
    let streamError: Extract<AskFrame, { type: "error" }> | undefined;
    for await (const frame of this) {
      if (frame.type === "event") {
        const payload = frame.payload as { delta?: string; text?: string } | undefined;
        text += payload?.delta ?? payload?.text ?? "";
      } else if (frame.type === "error") {
        streamError = frame;
      } else if (!frame.success) {
        throw new AgentStreamError(
          frame.error ?? streamError?.message ?? "agent run failed",
          streamError?.code ?? "internal",
          frame.upstreamStatus,
        );
      } else {
        return frame.text || text;
      }
    }
    return text;
  }
}

async function withTimeout<T>(promise: Promise<T>, timeoutMs: number): Promise<T> {
  let timer: ReturnType<typeof setTimeout> | undefined;
  try {
    return await Promise.race([
      promise,
      new Promise<never>((_, reject) => {