tf-data — collections: writes, reads, queries, vector search, REST¶

Every row carries a state, a user_id, and audit timestamps. The state drives pub/sub — downstream workers subscribe to state transitions via tf.on_state(collection, state).

Reserved collection names (do not write from agent code): _debug (stepped-debug toggle, see Stepped debugging below), _mcp_* (MCP server registrations). The leading underscore is the convention — keep your own collection names underscore-free.

Writes¶

# UPSERT by key. At least one of state= / data= is required. (Vectors are written
# separately via .set_vectors(key, items) — see *Vector writes* below.)
# On INSERT, omitted state defaults to 'new', omitted data defaults to {}.
# On UPDATE, each argument you pass replaces that column outright; arguments you
# omit are left untouched. There is NO field-level merge inside the JSONB:
# passing data= REPLACES the ENTIRE value blob (state= likewise replaces state).
tf.collection('documents').set('doc-123', state='loaded', data={'title': 'Q4 plan'})

# State-only transition: omit data= entirely to keep the existing value blob and
# only move the row's state. This is the one case that preserves the payload.
tf.collection('documents').set('doc-123', state='chunked')

# Data write: passing data= OVERWRITES the whole value. This drops every field
# not present in the dict you pass — 'title' here would wipe any sibling keys.
tf.collection('documents').set('doc-123', data={'title': 'Q4 plan v2'})

# INSERT a new row with an auto-generated UUID key. State is required.
new_key = tf.collection('chunks').add(state='new', data={'text': 'content'})

Whole-value replace — spread to update safely. data= is not a patch. Because it replaces the entire value JSONB, a handler that advances a row while setting one field MUST spread the existing payload, or every other field is nuked:

# CANONICAL safe-update idiom: spread the current data, then override.
tf.collection('documents').set(item['key'], state='processed',
                               data={**item['data'], 'analysed': True})

# WRONG — replaces the whole blob with a single key, destroying everything else.
tf.collection('documents').set(item['key'], state='processed',
                               data={'analysed': True})

The one documented exception is a state-only write (no data= at all), which preserves the existing value and only transitions state — no spread needed there.

tf.collection(...).set() and the REST PUT differ on data writes. The Python set(key, data=...) replaces the whole value blob (spread to preserve siblings, as above). The orchestrator's HTTP write path — PUT /api/factories/:name/data/:collection/:key (the composable-UI write path) — instead shallow-merges the JSONB (value = existing || patch, a top-level key merge; nested objects are still replaced wholesale, not deep-merged). So a UI patch of {"approved": true} keeps the row's other top-level fields, whereas the same one-key dict passed to set(data=...) would wipe them. State-only and state+data variants of the PUT are covered under Orchestrator REST surface below.

Reads — the lazy query builder¶

tf.collection(name) returns a lazy query builder. Filters chain and AND together; terminals execute. Nothing hits the DB until you call a terminal.

row      = tf.collection('documents').get('doc-123')             # full row dict, or None — point lookup
exists   = tf.collection('documents').exists('doc-123')          # bool — point lookup

all_rows = tf.collection('documents').get_all()                  # every row, newest first
loaded   = tf.collection('documents').state('loaded').get_all()  # filtered by state
n        = tf.collection('documents').state('loaded').count()    # int
ready_or_done = tf.collection('documents').state(['ready', 'done']).get_all()   # state IN (...)

Filters (chain, AND together):

Filter	Effect
`.state('X')`	`state = 'X'`
`.state(['X', 'Y', 'Z'])`	`state IN ('X', 'Y', 'Z')`
`.where("<DSL string>")`	a payload/column predicate (grammar below). Multiple `.where(...)` AND together.

Calling .state() twice raises ValueError — combine the values into one list instead of chaining two .state(...) calls.

Terminals (execute the query):

Terminal	Returns
`.get_all()` / `.run()` (alias)	`list[row dict]`, ordered `updated_at` descending — same row shape handlers receive
`.count()`	`int`
`.first()`	first row dict, or `None`
iteration (`for row in tf.collection(...).state(...)`)	iterates the rows

.vector_search(text_or_vec) is a filter, not a terminal — it sets ANN ordering and you still call a terminal (.run() / .get_all()) to execute. See Vector search below.

# chained: state filter + payload predicate + terminal
tf.collection('chunks').state('vectorised') \
    .where("document == 'ae400398.pdf' and token_count >= 400") \
    .get_all()

n = tf.collection('chunks').state(['vectorised', 'chunked']).count()

The `.where()` string DSL¶

.where("...") takes a small predicate language over a row's JSONB payload and (whitelisted) row columns. The string is parsed and parameterized — values are bound as SQL parameters, never concatenated into the query.

Operators: == != < > <= >= in "not in" and or not ( ). Literals: string (single- or double-quoted, no backslash escapes), number, bool (true/false), list [a, b, c].

Field namespaces — three ways to name a field, because the JSONB payload and the lifecycle columns can collide:

Reference form	Resolves to	Example
bare field	JSONB payload key (`value->>'field'`)	`document == 'x'`
`data.field`	explicit payload alias — same as bare, disambiguates	`data.state == 'VIC'`
`meta.<col>`	a row column (whitelisted)	`meta.created_at > '2026-01-01'`

meta.* whitelist: state, key, user_id, created_at, updated_at, state_changed_at. factory_name and collection are not addressable — they're pinned by the collection scope. Any unknown meta.* column, or any unknown namespace prefix, is a parse error.

The collision rule (this is why the namespaces exist). Consider an address collection where the payload has its own state field for the Australian state:

tf.collection('address').add(state='new', data={'suburb': 'Geelong', 'state': 'VIC'})

# bare `state` reads the PAYLOAD field:
tf.collection('address').where("state == 'VIC'").get_all()        # payload state == 'VIC'

# meta.state reads the LIFECYCLE column:
tf.collection('address').where("meta.state == 'new'").get_all()   # row's lifecycle state

Bare state → payload value->>'state'. The lifecycle column is always meta.state. (For filtering on the lifecycle state you'd normally use the .state(...) filter; meta.state exists for the rare case you want it inside a compound .where(...).)

Numeric / bool casting. Ordering operators (< > <= >=) and numeric/bool literals cast the JSONB text (::numeric / ::boolean) with a guard, so a row whose payload value isn't a valid number/bool is excluded from the result rather than erroring the whole query. String comparisons (==, in) don't cast. != is compiled to IS DISTINCT FROM (so a NULL payload field is correctly not equal to a value).

tf.collection('chunks').where("token_count >= 400 and token_count < 1200").get_all()
tf.collection('invoice').where("paid == true").count()

OR and grouping are supported:

tf.collection('lead').where("score >= 80 or (tier == 'gold' and active == true)").get_all()

Injection / untrusted-input note. Because .where() strings are parsed and parameterized (values bind as parameters, never string-concatenated into SQL), building a predicate with an f-string is safe against SQL injection:

name = item['data']['filename']
tf.collection('chunks').where(f"document == '{name}'").get_all()   # parameterized — injection-safe

Parse-safe is not the same as injection-safe. The interpolated value re-enters the DSL tokenizer, so a value containing a quote ', backslash \, or bracket [ ] raises QueryFilterError at .where() (a surfaced error, not swallowed) — even though the query would be SQL-injection-safe if it parsed. When you have the row key, prefer a keyed .get(key); otherwise validate the value before interpolating it.

However, if name is untrusted end-user input, parameterization only protects the SQL layer — confining a query to within-factory confidentiality (i.e. not letting a user craft a predicate that surfaces rows they shouldn't see) is the factory author's responsibility. Validate/scope untrusted predicate inputs yourself.

Not yet built (don't reach for these — they're deferred): .order_by(...), .min_similarity(...), per-group top-N (top_per), and .delete(). (or / grouping is built — it's listed above.)

Delete¶

tf.collection('chunks').remove('chunk-xyz')   # cascades to factory_vectors

There is no bulk-delete. If you need to delete many rows, iterate and call remove per key — that's intentional friction.

Vector writes — `set_vectors` (n:1)¶

Embeddings are n:1 with a factory_data row: one row owns many chunk-vectors (factory_vectors PK = factory_name, collection, key, chunk_index). Vectors are written with .set_vectors(key, items) — the one vector-write path. The parent row (via .set()/.add()) must exist first (FK).

# items = list of {'content': str, 'meta': dict} — or bare strings. ONE item is a
# row embedding; MANY are chunks. set_vectors embeds each content internally (one
# batched tf.embed call) and stores the content + per-chunk meta alongside the vector.
tf.collection('documents').set('AE530989.pdf', state='vectorised', data={'pages': 12})
tf.collection('documents').set_vectors('AE530989.pdf', [
    {'content': 'clause 3 — overtime rates …', 'meta': {'start_page': 3}},
    {'content': 'clause 8 — leave loading …',  'meta': {'start_page': 8}},
])

# A row embedding is a one-item list; a bare string works too.
tf.collection('people').set_vectors('sam@x.com', ['Sam — staff engineer, ML platform'])

Precomputed embeddings. Skip embedding cost by passing a precomputed vector: set_vectors accepts an optional embedding field per item. Items with precomputed vectors are stored as-is; only items without are embedded:

# Embedding is skipped for the first item.
tf.collection('documents').set_vectors('key', [
    {'content': 'pre-embedded text', 'meta': {...}, 'embedding': [0.1, 0.2, ...]},
    {'content': 'needs embedding', 'meta': {...}},  # embedded in one batch call
])

Chunk your text first with tf.chunk(...) (pure stdlib, provider-neutral — no tokenizer; sizes are in characters):

chunks = tf.chunk(doc).by_paragraphs()               # prose, paragraph-aware
chunks = tf.chunk(doc).by_chars(1000).overlap(100)   # fixed window + overlap
chunks = tf.chunk(md).by_markdown()                  # heading-sectioned; heading path in meta
tf.collection('documents').set_vectors(key, chunks)  # each chunk -> one vector

set_vectors is a full-key REPLACE, atomic (one transaction): every existing vector for key is deleted and the new set inserted — re-embedding never leaves orphans. set_vectors(key, []) clears the row's vectors. Removing the parent row cascades its vectors.

Embedding model. set_vectors embeds via the default provider/model; override per call with .set_vectors(key, items, provider='openrouter', model='baai/bge-m3'). The read side MUST use the same model — different models produce different dims + vector spaces, so a non-default write requires a matching read (see below). Dimension must match a fixed factory_vectors size — see tf-llm § Embeddings.

Vector search¶

.vector_search(text_or_vec) is a chainable filter, not a terminal — it sets ANN (cosine) ordering. Call a terminal (.run() / .get_all()) to execute and .limit(n) to cap (default 10). A string query is auto-embedded with the default model (which must match the write model); pass a pre-embedded vector to use another.

# Grouped default: the single best-matching chunk per parent row (deduped in SQL).
hits = tf.collection('documents').vector_search('budget overrun').limit(5).run()

# .chunks() — every matching chunk (within-a-document search).
# .key(k)  — restrict to one row's vectors.
hits = tf.collection('documents').vector_search('overtime') \
    .key('AE530989.pdf').chunks().limit(8).run()

# Object-level filters (.state()/.where()) narrow the PARENT rows first.
hits = tf.collection('documents').where("state_field == 'VIC'") \
    .vector_search('leave loading').limit(5).run()

# Non-default embedding model: embed the query with the SAME model.
q = tf.embed('budget overrun').provider('openrouter').model('baai/bge-m3')
hits = tf.collection('documents').vector_search(q).limit(5).run()

# Each hit is the parent row dict + `similarity` (cosine 0..1) + `chunk`:
#   {factory_name, collection, key, ..., similarity,
#    chunk: {content, index, meta}}

State filtering via the builder¶

Reads filter by state via the .state(...) filter — the state= keyword argument is never used:

tf.collection('c').state('x').get_all()
tf.collection('c').state('x').count()
tf.collection('c').state('x').vector_search(q).limit(5).run()

The user_id column defaults to 'system' for agent writes. Backend API writes stamp the session's user id (fallback '-1' until auth is wired).

Orchestrator REST surface for `factory_data`¶

The orchestrator exposes the same factory_data rows over HTTP for UI consumers. Agent-side Python code uses tf.collection(...) directly; UI code uses these endpoints (and is the canonical client of useBoundData).

Method	Path	Purpose
`GET`	`/api/factories/:name/data/:collection`	List rows in a collection. Paginated, sortable, filterable by state, filterable by recency via `?since=`.
`GET`	`/api/factories/:name/data/:collection/stats`	Aggregate stats (group-by JSONB field, or daily counts over N days).
`GET`	`/api/factories/:name/data/:collection/:key`	Single row by key.
`PUT`	`/api/factories/:name/data/:collection/:key`	Upsert one row. Body: `{data?, state?}` (at least one required). Shallow-merges `data`.
`DELETE`	`/api/factories/:name/data/:collection/:key`	Delete one row by key.
`GET`	`/api/factories/:name/_state_counts`	`[{collection, state, count}]` across all user-facing collections (excludes `_`-prefixed reserved collections).

`PUT /api/factories/:name/data/:collection/:key` — upsert a row (shallow-merge)¶

The HTTP write path used by the composable UI. Body must include data, state, or both (else 400). This is the one write path that does NOT replace the whole value blob — on an existing row it shallow-merges the JSONB (value = existing || patch, a top-level key merge):

Body	On new row	On existing row
`{data, state}`	inserts `value = data`, `state = state`	`value = existing ‖ data` (top-level merge), `state` overwritten
`{data}` only	inserts `value = data`, `state = 'new'`	`value = existing ‖ data`; state preserved
`{state}` only	inserts `value = {}`, `state = state`	`state` overwritten; value preserved

‖ is a top-level merge (Postgres jsonb ||): keys in data overwrite same-named top-level keys; keys absent from data survive; a nested object under a shared key is replaced wholesale (no deep merge). Every PUT stamps user_id from the session. Contrast the Python surface: tf.collection(...).set(key, data=...) replaces the whole blob (spread to preserve siblings).

`GET /api/factories/:name/data/:collection` — list rows¶

Query params

Param	Type	Default	Notes
`state`	string	(none)	Filter rows by state.
`page`	int ≥ 1	`1`	Ignored when `since` is set.
`page_size`	int 1..1000	`50`	Ignored when `since` is set.
`sort_field`	JSONB key matching `[a-zA-Z_][a-zA-Z0-9_]{0,63}`	`updated_at`	Strict allowlist — anything else 400s.
`sort_dir`	`asc` \| `desc`	`desc`	Forced to `asc` when `since` is set (chronological catch-up).
`since`	ISO-8601 timestamp	(none)	Returns rows where `updated_at > since`. See below.

Default response shape (no since)

{
  "rows": [{ "key": "...", "data": {...}, "state": "...", "user_id": "...", "created_at": "...", "updated_at": "..." }],
  "page": 1,
  "page_size": 50,
  "total": 312,
  "total_pages": 7
}

?since=<ISO-8601> — reconnect catch-up

The composable useBoundData hook uses this to recover from a WebSocket disconnect without throwing away cached state. Caller records the largest updated_at it has seen, then on reconnect issues ?since=<that timestamp> to pick up only rows that have changed since.

Behaviour:

Rows are filtered by updated_at > since (strict greater-than — caller passes the last value it has, server returns everything strictly newer).
updated_at is bumped on every INSERT and every UPDATE (state transition or data write — the PUT shallow-merges the JSONB rather than replacing it, but any change still bumps updated_at), so since catches state-only transitions on existing rows. created_at is not used.
AND-composes with ?state=. Both filters apply.
Rows are returned ordered by updated_at ASC (oldest first within the window) so the caller can advance its cursor monotonically.
page and page_size are ignored. A backstop hard cap of 10000 rows applies; if the cap is hit, truncated: true is returned and the caller should fall back to a full re-fetch (request without since).
The cutoff is echoed back as since: "<iso>" so the caller can confirm the server honoured it.

Response shape with ?since

{
  "rows": [...],
  "since": "2026-05-03T12:34:56.000Z",
  "total": 17,
  "truncated": false
}

page / page_size / total_pages are omitted from the since-response. total is rows.length.

Error path

Non-parseable since value → 400 {"error": "bad_request", "detail": "since must be ISO-8601 (e.g. 2026-05-03T12:34:56Z)"}. Validation is const t = new Date(since); if (isNaN(t.getTime())) reject.
Future-dated since is allowed (returns empty rows: []); server does not clamp.

Tombstones / deletes — current constraint

Rows deleted between disconnect and reconnect do not appear in ?since= results — the caller has no way to know to evict them. Mitigation: tf_data_changed carries op: 'delete' for live deletes, so an open SSE connection sees them; on reconnect, the next NOTIFY on the collection (any cause) prompts useBoundData to refetch, which catches deletions that happened during the gap. A full deletion-replay (e.g. a soft-delete tombstone column queried by ?since=) is deferred — the live channel covers the common case.

Backwards compat

Purely additive. Callers without ?since= get the existing page/page_size/total_pages shape unchanged.

tf-data — collections: writes, reads, queries, vector search, REST¶

Writes¶

Reads — the lazy query builder¶

The .where() string DSL¶

Delete¶

Vector writes — set_vectors (n:1)¶