Function streamSessionItems

services/session-ingest/src/queue-consumer.ts:256–296

(
  r2Key: string,
  body: ReadableStream<Uint8Array>,
  onItem: (rawItem: Record<string, unknown>) => Promise<void>
)

      return;
    }
    for (const change of ingestResult.changes) {
      mergedChanges.set(change.name, change.value);
    }
    mergedAttentionSignals.push(...(ingestResult.attentionSignals ?? []));
  };

  const stage = async (rawItem: Record<string, unknown>): Promise<boolean> => {
    if (!accepted) return false;
    const parsed = SessionItemSchema.safeParse(rawItem);
    if (!parsed.success) {
      console.warn('Skipping invalid item in queue consumer', {
        r2Key,
        type: rawItem['type'],
        errors: parsed.error.issues.map(i => i.message),
      });
      return true;
    }

    const item = parsed.data;
    const { item_id } = getItemIdentity(item);

    const itemDataJson = JSON.stringify(item.data);
    const itemDataBytes = encoder.encode(itemDataJson).byteLength;

    if (chunkItemIds.has(item_id)) {
      await flushChunkToSessionDO();
      if (!accepted) return false;
    }

    // Offload data above the DO SQLite row limit to R2; the DO stores a
    // reference and an empty inline blob. Send only identity fields over RPC so
    // a single oversized item cannot exceed Cloudflare's RPC payload limit.
    const itemForRpc = itemDataBytes > MAX_INGEST_ITEM_BYTES ? slimItemForR2Reference(item) : item;
    const itemForRpcDataBytes =
      itemDataBytes > MAX_INGEST_ITEM_BYTES
        ? encoder.encode(JSON.stringify(itemForRpc.data)).byteLength
        : itemDataBytes;
    if (itemDataBytes > MAX_INGEST_ITEM_BYTES) {
      const itemR2Key = `items/${kiloUserId}/${sessionId}/${item_id}/${ingestedAt}`;
      const staged = await withDORetry(
        () => getSessionIngestDO(env, { kiloUserId, sessionId }),
        stub =>
          stub.stageR2Object(
            { kiloUserId, sessionId, key: itemR2Key },