Streams

A stream is a job that takes its items one push at a time and reports each item, once it finalizes, on an ordered results feed. For the concepts and a worked loop, see Online RL with Streams.

Using online RL for your business? Contact us before you start, so we can allocate rater bandwidth for your training runs.

Every endpoint below is under https://api.trydatapoint.com/data-labelling/v1, takes your API key in the X-API-Key header and counts against your key's rate limit. A request without a valid key is answered 401 before any stream lookup, whatever its stream_id; only a body that isn't valid JSON is refused before that, with a 422. A stream's stream_id is also a job id: streams appear in GET /jobs, and the endpoints below answer 404 for an ordinary job's id. Of the job endpoints, POST /jobs/{stream_id}/pause and /resume work on a stream; /cancel and /complete answer 400 with code use_stream_close (end a stream with close instead), and /retry answers 400.

Create a stream

Creates a stream, or returns your existing stream with this name when its settings match, so a retried create is safe.

Request body

Targeting lists the filters annotator_filter takes.

task_type, instruction, response_options, max_responses_per_item, annotator_filter and serving_environment are fixed for the stream's life. Creating a stream again under the same name with any of them different is refused with 409 config_mismatch, whose fields names them. item_ttl_seconds, max_pending_items and budget_credits apply only when the stream is new: an existing stream keeps its own, and PATCH changes them.

Streams don't take chain steps, dimensions, custom task UIs or datapoints. An unknown field is a 422. So is invalid text (a NUL character or an unpaired surrogate) in name, instruction, response_options or annotator_filter, and NaN or Infinity in response_options. A name held by one of your ordinary jobs is a 409 name_conflict.

The existing stream comes back whatever its status, so creating dpo-run-1 again after closing it returns the closed stream, and pushes to it get 409 stream_closed. Use a new name for each run.

Response

The stream object.

curl -X POST https://api.trydatapoint.com/data-labelling/v1/streams \
  -H "X-API-Key: $DATAPOINT_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "dpo-run-1",
    "instruction": "Which image better matches the prompt? {context}",
    "task_type": "comparison",
    "max_responses_per_item": 3,
    "serving_environment": "sandbox",
    "item_ttl_seconds": 120
  }'

Push items

Adds 1 to 500 items to an open stream. Each one is served to raters and finalizes by its deadline.

Request body

Task types gives the media shape of each task type.

What happens to a push

The checks run in this order, and a refused push stores nothing:

0. The request itself. Before any of the checks below, the body must be valid JSON (422), and the request must carry a valid API key (401), be within your key's rate limit (429, with Retry-After) and match the request schema (422: no items or more than 500, an out-of-range ttl_seconds, a blank or over-long external_id, oversized metadata, invalid text or a NaN (see the item rules), an unknown field).
1. The stream must be yours. Otherwise 404.
2. No external_id may repeat within the push. Otherwise 400 invalid_items, naming each repeat.
3. A stored external_id must keep its content. Pushing a known external_id with the same context and media returns the stored item with duplicate: true and charges nothing. Different content is 409 external_id_conflict, whatever state the stream is in.
4. New items need an open, unpaused stream. A push with any new item is refused with 409 stream_closed on a closed or closing stream and 409 stream_paused on a paused one. A stream is paused by you (POST /jobs/{stream_id}/pause, or Pause on its dashboard page) or by our team; resume it with POST /jobs/{stream_id}/resume. Pending items still finalize at their deadlines while it is paused. A push whose items are all already stored is answered with their duplicates even then, so a retry always learns what was stored.
5. New items must fit under max_pending_items. More new items than the cap could never fit, so each one past it is a 400 invalid_items. Otherwise, if pending plus new items would pass the cap, the push is a 429 stream_backpressure with a Retry-After.
6. Every new item must be valid. Otherwise 400 invalid_items, listing every bad item (see the item rules below).
7. Text review, which is off unless enabled for your account (contact us to turn it on). Text that doesn't pass is a 422 content_blocked, naming the items in blocked_items. Items are reviewed in groups, so blocked_items can include an item whose own text is fine; split the push to narrow it down. Media is reviewed, whether uploaded or fetched from a URL; an item whose media is blocked finalizes blocked.
8. Credits. 402 stream_budget_exceeded past budget_credits, 409 stream_limit_reached past the stream's lifetime limit on reserved credits, and 402 when your balance can't cover the push.

A push can wait behind other updates to your account's streams and jobs. One that waited too long is answered 503 stream_busy with Retry-After: 1. It stored nothing: retry it.

Item rules

• context reaches raters only through {context} in the stream's instruction; for multiple_choice it is the question itself. An item with context on a stream whose instruction has no {context} is refused, except on multiple_choice.
• multiple_choice and text_answer items take only uploaded media (dp:// refs or media_id) in their media roles.
• A pairwise_ranking item needs a max_responses_per_item of at least its number of candidates (1 for two candidates).
• Option on_select actions are for chain jobs only.
• external_id, context and every string in metadata and media, keys included, must be valid text: no NUL character (\u0000) and no unpaired surrogate (such as \ud800). metadata and media can't hold NaN or Infinity. Any of these is a 422 whose field error names the item's position in items.
• When the push breaks a rule that no single item explains, it is refused as a whole: one error, on the first new item, reading "this push is invalid as a whole: …".

Response (200 OK)

{
  "stream_id": "job_5f2c9a7e1b3d",
  "items": [
    { "index": 0, "external_id": "dpo-run-1-0-0", "duplicate": false },
    { "index": 1, "external_id": "dpo-run-1-0-1", "duplicate": false }
  ],
  "pending_items": 2,
  "reserved_credits": 0
}

dpo-run-1 is a sandbox stream, so its rate is 0 and the push holds nothing. On a prod stream with a rate of 5 credits per vote, the same push would hold 2 items × 3 votes × 5 = 30 credits.

Items whose media is ready, such as text or dp:// uploads that passed review, can be served as soon as the push returns. Others, such as public http(s) URLs, are fetched and checked by background processing first, which adds latency.

curl -X POST https://api.trydatapoint.com/data-labelling/v1/streams/$STREAM_ID/items \
  -H "X-API-Key: $DATAPOINT_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "items": [
      {
        "external_id": "dpo-run-1-0-0",
        "context": "a warm sunset",
        "media": {"candidates": [{"url": "dp://b6a1cd3f1234/a.png"}, {"url": "dp://c8f3ae9b5678/b.png"}]},
        "metadata": {"step": 0}
      },
      {
        "external_id": "dpo-run-1-0-1",
        "context": "a city at night",
        "media": {"candidates": [{"url": "dp://d2e7bf0a9012/c.png"}, {"url": "dp://e4a9c1d03456/d.png"}]},
        "metadata": {"step": 0}
      }
    ]
  }'

Read results

Returns finalized items with a seq greater than after, in the order they finalized. Each finalized item gets the next seq, with no gaps, so paging with after sees each item exactly once.

Query parameters

Response (200 OK)

{
  "stream_id": "job_8d41e07c2a96",
  "results": [
    {
      "seq": 1,
      "index": 0,
      "external_id": "grpo-run-7-0-0",
      "metadata": null,
      "outcome": "completed",
      "pushed_at": "2026-10-20T14:03:11.204000Z",
      "expires_at": "2026-10-20T14:05:11.204000Z",
      "finalized_at": "2026-10-20T14:03:52.917000Z",
      "error": null,
      "ingested": true,
      "result": {
        "total_responses": 6,
        "comparisons": { "0": 3, "1": 3, "2": 3, "3": 3 },
        "wins": [[0, 1, 1, 1], [0, 0, 1, 1], [0, 0, 0, 1], [0, 0, 0, 0]],
        "win_rate": { "0": 1.0, "1": 0.667, "2": 0.333, "3": 0.0 },
        "scores": { "0": 1.2264, "1": 0.3783, "2": -0.3783, "3": -1.2264 },
        "ranking_order": ["0", "1", "2", "3"],
        "weighted_wins": [[0.0, 1.0, 1.0, 1.0], [0.0, 0.0, 1.0, 1.0], [0.0, 0.0, 0.0, 1.0], [0.0, 0.0, 0.0, 0.0]],
        "weighted_scores": { "0": 1.2264, "1": 0.3783, "2": -0.3783, "3": -1.2264 },
        "weighted_ranking_order": ["0", "1", "2", "3"]
      },
      "responses": null
    }
  ],
  "next_after": 1,
  "has_more": false
}

Item lifecycle explains how an item reaches each outcome.

• With wait_seconds, the request returns once min_items items are ready, once a page is full when a full page holds fewer than min_items, or when the wait runs out. Each long-poll counts as one request against your rate limit; while items keep finalizing, a loop that asks again at once costs one request per batch, so raise min_items or space the requests out.
• Keep reading with after set to next_after while has_more is true. A page holds at most limit items, and fewer on a stream with a high max_responses_per_item.
• A failed or blocked item reports the votes it received before it failed or was blocked, but it isn't billed for them (credits.spent excludes them). Train only on completed and expired items.
• An item's result never changes once it has finalized.
• An item can finalize completed with fewer votes than max_responses_per_item when quality checks discarded some of its answers. total_responses is the number that counts.
• An item whose media was never stored, such as one that expired before it was ingested, has total_responses 0 and the task type's empty aggregate. For pairwise ranking that has no candidate positions: comparisons, win_rate, scores and weighted_scores are {}, and wins and weighted_wins are [].
• This endpoint never returns 402.

Get a stream

Returns the stream object, with recent latency figures.

Change a stream's limits

Send only the fields you want to change, at least one. The new values apply to later pushes: items already pushed keep their deadlines, and a max_pending_items below what is already pending refuses new pushes until enough items finalize. Finalizing gives back unused holds but never lowers what the stream has spent, so a budget_credits below credits.spent stays exceeded until you raise it. A paused stream can be changed.

Returns the stream object, with every latency_seconds field null; GET /streams/{stream_id} has the figures. A closed or closing stream answers 409 stream_closed. A change that waited too long behind another update to the stream is answered 503 stream_busy with Retry-After: 1; it changed nothing, so send it again.

Expire items

Finalizes named pending items now, each with the votes it has. Name them by external_ids or by indices, not both, 1 to 500 per request.

{ "external_ids": ["dpo-run-1-0-0", "dpo-run-1-0-1"] }

Response: {"expired": 2}, the number of named items that were still pending; unknown and already finalized items are skipped.

• An item that never reached raters expires with no votes; one that already had all its votes finalizes completed.
• An item whose media failed or was blocked finalizes failed or blocked, not expired.
• Some named items can take a few seconds after the request returns to finalize; they reach the results feed as usual.
• These are a 422: both fields or neither; no names or more than 500; an empty string in external_ids, or one holding a NUL character or an unpaired surrogate; an index that isn't an integer (true, "3" and 2.0 aren't) or is outside 0 to 2,147,483,647.
• A request that waited too long behind another update to the stream is answered 503 stream_busy with Retry-After: 1. It expired nothing: send it again.
• Expiring is safe to retry, but expired counts only the items still pending when that request ran: a retry after a lost answer reports 0 for the items the first attempt expired.

Close a stream

Stops the stream taking new items. The body is optional:

{ "drain": true }

With drain: true (the default), pending items keep collecting votes until they complete or reach their deadlines; with drain: false they expire now. Closing again is safe, and drain can change from true to false, not back.

Returns the stream object, with status closing while items are pending and closed once none are. With drain: false the answer can still say closing for a few seconds before the stream reports closed. A close that waited too long behind another update to the stream is answered 503 stream_busy with Retry-After: 1; it changed nothing, so send it again. A stream closes itself, without draining, 14 days after it was created. After closing, read the results feed until it is drained.

The stream object

{
  "stream_id": "job_8d41e07c2a96",
  "name": "grpo-run-7",
  "status": "open",
  "task_type": "pairwise_ranking",
  "instruction": "Which reply better answers the prompt? {context}",
  "max_responses_per_item": 6,
  "item_ttl_seconds": 120,
  "max_pending_items": 1000,
  "budget_credits": null,
  "serving_environment": "sandbox",
  "annotator_filter": null,
  "response_options": null,
  "is_paused": false,
  "items": { "pushed": 40, "pending": 12, "completed": 25, "expired": 3, "failed": 0, "blocked": 0 },
  "credits": { "rate": 0, "reserved": 0, "spent": 0 },
  "latency_seconds": { "ingest_p50": 0.412, "finalize_p50": 41.37, "finalize_p90": 88.912 },
  "last_seq": 28,
  "created_at": "2026-10-20T14:01:07.551000Z",
  "closed_at": null
}

The other fields echo the stream's settings.

Errors

Error bodies use the site-wide error envelope. Its detail is an object with a code for most stream errors (the balance 402 is an object without one), a string for the rest, and an array of field errors for a schema 422.

Retrying safely

• Give every item an external_id, and make it deterministic, such as {run}-{step}-{i}, so a restarted process pushes the same ids.
• A refused push stores nothing. Fix it and send it again; after a 429 or 503, wait Retry-After and send the same request.
• After a timeout or a 5xx, send the same items with the same external_ids. Stored items come back with duplicate: true and aren't charged again, even if the stream was paused or closed in between.
• Items without an external_id can't be matched this way: a resent copy is a new item, charged again.
• New content under a stored external_id is refused (409 external_id_conflict). Each upload of a file gets a new dp:// ref, so upload once and reuse the ref when you resend.

Media

Upload files with POST /media, several to a request, and push their dp:// refs. Public http(s) URLs are fetched and checked by background processing, which adds latency before raters see the item. multiple_choice and text_answer items take only uploaded media.