local stream_ready=false
local max_reconnects="\\${ARK_MAX_SSE_RECONNECTS:-3}"
local max_history_pages="\\${ARK_MAX_HISTORY_PAGES:-100}"
: "\\${ARK_BASE_URL:?Set ARK_BASE_URL before calling run_session_turn}"
: "\\${ARK_API_KEY:?Set ARK_API_KEY before calling run_session_turn}"
: "\\${session_id:?Pass a Session ID as the first argument}"
: "\\${user_text:?Pass a user message as the second argument}"
command -v curl >/dev/null || {
printf 'curl is required.\n' >&2
command -v jq >/dev/null || {
printf 'jq is required.\n' >&2
[[ $max_reconnects =~ ^[0-9]+$ ]] || {
printf 'ARK_MAX_SSE_RECONNECTS must be a non-negative integer.\n' >&2
[[ $max_history_pages =~ ^[1-9][0-9]*$ ]] || {
printf 'ARK_MAX_HISTORY_PAGES must be a positive integer.\n' >&2
stream_dir=$(mktemp -d) || {
printf 'Failed to create a temporary directory for the SSE stream.\n' >&2
stream_error="$stream_dir/stream-error.log"
seen_ids_file="$stream_dir/seen-event-ids"
if [[ -n $stream_pid ]] && kill -0 "$stream_pid" 2>/dev/null; then
kill "$stream_pid" 2>/dev/null || true
if [[ -n $stream_pid ]]; then
wait "$stream_pid" 2>/dev/null || true
if [[ -n $stream_pipe ]]; then
trap cleanup_stream EXIT
stream_number=$((stream_number + 1))
stream_pipe="$stream_dir/events-$stream_number"
if ! mkfifo "$stream_pipe"; then
printf 'Failed to create the SSE pipe at %s.\n' "$stream_pipe" >&2
curl -sS -N --fail-with-body --max-time 600 \
"$ARK_BASE_URL/sessions/$session_id/events/stream" \
-H "Authorization: Bearer $ARK_API_KEY" \
-H "Accept: text/event-stream" >"$stream_pipe" 2>>"$stream_error" &
while IFS= read -r line <&3; do
if [[ $line == ": ready" ]]; then
if [[ $stream_ready != true ]]; then
printf 'SSE stream closed before it was ready.\n' >&2
if [[ -s $stream_error ]]; then
tail -n 1 "$stream_error" >&2
[[ -n $event_id ]] && grep -Fqx -- "$event_id" "$seen_ids_file"
if [[ -n $event_id ]] && ! event_seen "$event_id"; then
printf '%s\n' "$event_id" >>"$seen_ids_file"
event_id=$(jq -r '.id // empty' <<<"$event_json" 2>/dev/null || true)
if event_seen "$event_id"; then
remember_event "$event_id"
event_type=$(jq -r '.type // empty' <<<"$event_json" 2>/dev/null || true)
jq -r '.content[]? | select(.type == "text") | .text' <<<"$event_json"
message_count=$((message_count + 1))
agent.tool_use | agent.mcp_tool_use | agent.custom_tool_use)
printf '[tool] %s\n' "$(jq -r '.name // .type' <<<"$event_json")"
retry_status=$(jq -r '.error.retry_status.type // empty' <<<"$event_json")
printf '[error] %s: %s\n' \
"$(jq -r '.error.type // "unknown_error"' <<<"$event_json")" \
"$(jq -r '.error.message // "No error message returned."' <<<"$event_json")" >&2
printf '[session] The service is retrying automatically.\n' >&2
printf '[session] Retries were exhausted for this turn.\n' >&2
printf '[session] The Session cannot continue after this error.\n' >&2
printf '[session] Unknown error retry status: %s\n' "$retry_status" >&2
session.status_rescheduled)
printf '[session] The Session was rescheduled; continue waiting.\n' >&2
stop_reason=$(jq -r '.stop_reason.type // empty' <<<"$event_json")
if ((message_count == 0)); then
printf 'The turn ended without an agent.message event.\n' >&2
printf '[session] end_turn\n' >&2
pending_ids=$(jq -r '.stop_reason.event_ids // [] | join(",")' <<<"$event_json")
printf '[session] requires_action: %s\n' "$pending_ids" >&2
printf '[session] retries_exhausted\n' >&2
printf 'Unknown session.status_idle stop_reason: %s\n' "$stop_reason" >&2
printf '[session] deleted\n' >&2
session.status_terminated)
printf '[session] terminated\n' >&2
page_count=$((page_count + 1))
if ((page_count > max_history_pages)); then
printf 'History recovery exceeded %s pages.\n' "$max_history_pages" >&2
response=$(curl -sS --fail-with-body --get \
"$ARK_BASE_URL/sessions/$session_id/events" \
-H "Authorization: Bearer $ARK_API_KEY" \
--data-urlencode "limit=200" \
--data-urlencode "order=asc" \
--data-urlencode "page=$page") || return 1
response=$(curl -sS --fail-with-body --get \
"$ARK_BASE_URL/sessions/$session_id/events" \
-H "Authorization: Bearer $ARK_API_KEY" \
--data-urlencode "limit=200" \
--data-urlencode "order=asc") || return 1
while IFS= read -r event_json; do
[[ -n $event_json ]] || continue
if [[ $mode == "seed" ]]; then
remember_event "$(jq -r '.id // empty' <<<"$event_json")"
process_event "$event_json"
[[ $turn_done == true ]] && break
done < <(jq -c '.data[]?' <<<"$response")
[[ $turn_done == true ]] && return 0
page=$(jq -r '.next_page // empty' <<<"$response")
[[ -n $page ]] || return 0
if ! read_history seed; then
printf 'Failed to read the Session history before sending the message.\n' >&2
events_payload=$(jq -n --arg text "$user_text" \
'{events: [{type: "user.message", content: [{type: "text", text: $text}]}]}')
curl -sS --fail-with-body \
"$ARK_BASE_URL/sessions/$session_id/events" \
-H "Authorization: Bearer $ARK_API_KEY" \
-H "Content-Type: application/json" \
-d "$events_payload" >/dev/null || return 1
while [[ $turn_done != true ]]; do
while IFS= read -r line <&3; do
if [[ $line == ": server draining, reconnect" ]]; then
[[ $line == data:* ]] || continue
[[ $turn_done == true ]] && break
[[ $turn_done == true ]] && break
if ((reconnect_count >= max_reconnects)); then
printf 'SSE reconnect limit reached before the turn finished.\n' >&2
reconnect_count=$((reconnect_count + 1))
printf '[session] Reconnecting SSE and recovering history (%s/%s).\n' \
"$reconnect_count" "$max_reconnects" >&2
if ! read_history recover; then
printf 'Failed to recover Session history after reconnecting.\n' >&2