Stream a response token by token

#!/usr/bin/env bash
# Recipe: stream-a-response
#
# Consumes /v4/interact/stream, which answers with server-sent events rather
# than JSON: one `event: trace` frame per trace, then a final `event: end`.
# Proves the frames arrive incrementally by stamping each one with the second
# it was read in, and reassembles the agent's reply from the token deltas.
#
# Required environment:
#   VF_API_KEY     project API key (Settings -> API keys), VF.DM....
#   VF_PROJECT_ID  the project ID (Settings -> General -> Metadata)
set -euo pipefail

: "${VF_API_KEY:?set VF_API_KEY to a project API key}"
: "${VF_PROJECT_ID:?set VF_PROJECT_ID to the project ID}"

BASE="https://general-runtime.voiceflow.com"
USER_ID="cookbook-$(date +%s)"
WORK=$(mktemp -d)
trap 'rm -rf "$WORK"' EXIT

# The runtime can emit raw control characters inside trace payload strings,
# which strict JSON parsers reject. Strip them before jq.
json() { tr -d '\000-\037'; }

fail() { echo "FAIL at $1" >&2; exit 1; }

# 1. Start a session. Authenticates with the PROJECT API KEY.
SESSION=$(curl -sf -X POST "$BASE/v4/project/$VF_PROJECT_ID/session" \
  -H "Authorization: $VF_API_KEY" \
  -H "Content-Type: application/json" \
  -d "{ \"userID\": \"$USER_ID\" }") || fail "start session (a 401 here means the API key is wrong)"
SESSION_KEY=$(printf '%s' "$SESSION" | json | jq -er '.sessionKey') || fail "no sessionKey in the response"

# 2. Launch the conversation, streamed. Every call below authenticates with
#    the SESSION KEY, never the API key.
curl -sfN -X POST "$BASE/v4/interact/stream" \
  -H "Authorization: $SESSION_KEY" \
  -H "Content-Type: application/json" \
  -d '{ "action": { "type": "launch" } }' > "$WORK/launch" \
  || fail "launch (a 401 here means the API key was sent where the session key belongs)"
grep -q '^event: end$' "$WORK/launch" || fail "the launch stream never sent its end event"

# 3. Send a user message. `completionEvents` asks the runtime to emit the
#    generated text token by token instead of one finished text trace.
#
#    Read the body line by line as it arrives, stamping each line with the
#    seconds elapsed since the request went out. SECONDS is a shell builtin,
#    so the stamp costs no subprocess and cannot itself delay the read.
SECONDS=0
curl -sfN -D "$WORK/headers" -X POST "$BASE/v4/interact/stream" \
  -H "Authorization: $SESSION_KEY" \
  -H "Content-Type: application/json" \
  -d '{
        "action": { "type": "text", "payload": "In a few sentences, what can you help me with?" },
        "config": { "completionEvents": true }
      }' \
  | while IFS= read -r line; do printf '%s %s\n' "$SECONDS" "$line"; done > "$WORK/turn" \
  || fail "stream the turn (a 401 here means the API key was sent where the session key belongs)"

grep -qi '^content-type: text/event-stream' "$WORK/headers" \
  || fail "the response was not an event stream (check the Content-Type header)"

# Each frame is three lines - `event: <name>`, `id: <n>`, `data: <json>` -
# followed by a blank line. Split the stamp back off before parsing.
BODY=$(cut -d' ' -f2- < "$WORK/turn")
EVENTS=$(printf '%s\n' "$BODY" | sed -n 's/^event: //p')
FRAMES=$(printf '%s\n' "$BODY" | sed -n 's/^data: //p' | json | jq -s '.') \
  || fail "a data: line was not a complete JSON document"

printf '%s\n' "$EVENTS" | grep -qx 'trace' || fail "the stream carried no trace events"
printf '%s\n' "$EVENTS" | tail -1 | grep -qx 'end' || fail "the stream did not terminate with an end event"

# The generated text arrives as completion traces: one `start`, many
# `content` deltas, one `end`. Reassembling them is the whole point.
printf '%s' "$FRAMES" | jq -e 'map(select(.type == "completion") | .payload.state) | index("start") != null and index("end") != null' > /dev/null \
  || fail "no completion trace pair (state start .. end) in the stream"
DELTAS=$(printf '%s' "$FRAMES" | jq '[.[] | select(.type == "completion" and .payload.state == "content")] | length')
[ "$DELTAS" -gt 1 ] || fail "expected many content deltas, got $DELTAS (is completionEvents set in the request BODY?)"
REPLY=$(printf '%s' "$FRAMES" | jq -rj '.[] | select(.type == "completion" and .payload.state == "content") | .payload.content')
[ -n "$REPLY" ] || fail "the reassembled reply was empty"

# A turn that exhausted the workspace's credits reports it as a debug trace,
# not as a transport error. Catch it here rather than shipping a blank reply.
if printf '%s' "$FRAMES" | jq -e 'any(.[]; tostring | test("credit_limit_reached"))' > /dev/null; then
  fail "the turn hit the workspace credit limit"
fi

# Incremental delivery is the claim this endpoint makes. If the response were
# buffered, every frame would be read in the same second.
FIRST=$(head -1 "$WORK/turn" | cut -d' ' -f1)
LAST=$(tail -1 "$WORK/turn" | cut -d' ' -f1)
SPREAD=$((LAST - FIRST))
[ "$SPREAD" -ge 1 ] || fail "every frame arrived in the same second, which is what a buffered response looks like"

echo "ok: $(printf '%s\n' "$EVENTS" | grep -cx trace) trace event(s) over ${SPREAD}s, $DELTAS content deltas reassembling to ${#REPLY} characters, types: $(printf '%s' "$FRAMES" | jq -c '[.[] | .type // empty] | unique')"