Data & Integration

The top-level body is a straight sequence

Everything on this page — query, command, parallel over — is a top-level operation. None of them can appear inside an agent method or a fn body. The parser refuses by name:

Error: 'query' can only be used as a top-level operation, not in an agent method or fn body.
Error: Unrecognized step in Worker.go (line 9): results = parallel over rows as row limit 2 -> Out[]

This is worth stating plainly rather than discovering: you cannot query conditionally, inside a loop, or from a value you derived in a method. The data flow of a program is a flat sequence at the top level, and the branching lives in what you do with the rows after they arrive.

The shape this produces

A pipeline step therefore reads top to bottom, and the model work sits in the middle of it:

read the watermark      state.read("orders.since")
        │
pull rows               query Orders(since)          ← validated against Order
        │
reshape                 col.sortBy · col.groupBy · col.chunk
        │
the model               parallel over rows … { agent.classify(row) }
        │
write back              command MarkExported(since, batch)
        │
advance the watermark   state.write("orders.since", newest)

Conditional logic that would have wrapped a query goes one of two places instead: into the SQL as a parameter, or into an agent method that receives the rows. Both are usually the better design anyway — the first keeps the filtering where the index is, the second keeps the decision testable.

datasource · query · command

Luma owns the typed contract, not the driver. A database driver is an npm package, and shipping one would break the transpiler's single-file, zero-dependency promise — so the language describes the boundary and you supply roughly twenty lines of driver code.

datasource Warehouse {
  credential: "PG_URL"
  pool_max: 4
  timeout: 30000
  retry: 2
}

schema Order {
  id: string
  amount: float
  created_at: string @fmt("date-time")
}

query Orders(since: string) -> Order[] {
  source: Warehouse
  sql: "select id, amount, created_at from orders where created_at > $1 order by created_at"
}

command MarkExported(since: string, batch: string) -> int {
  source: Warehouse
  sql: "update orders set export_batch = $2 where created_at > $1"
  side_effects: "write"
}

rows     = query Orders("2026-01-01T00:00:00Z")
exported = command MarkExported("2026-01-01T00:00:00Z", "batch-1")

Declaration fields

The bridge is yours

A tool module exports the connections under a known name. This is the only part of the boundary Luma does not own:

export const datasources = {
  Warehouse: async (sql, params, signal) => {
    const result = await pool.query(sql, params);   // pg, mysql2, better-sqlite3 — your choice
    return result.rows;
  },
};

Without one, the program stops before it runs anything: Error: datasource 'Warehouse' has no registered bridge; export it from a tool module as datasources.Warehouse.

One row, or many

A return type of Schema[] takes whatever comes back. A bare Schema asserts exactly one row and fails otherwise, which is what you want for an aggregate:

schema OrderTotal {
  counted: int
  summed: float
}

query Totals(since: string) -> OrderTotal {
  source: Warehouse
  sql: "select count(*) as counted, sum(amount) as summed from orders where created_at > $1"
}

Error: query 'Totals' expected exactly one row, got 3.

Error: Datasource 'Warehouse' does not support 'transaction'; it is deliberately out of this package — handle transactions inside the bridge (ROADMAP § U4).

HTTP

std/http reaches services that are not databases — an API, a webhook, an object store with an HTTP interface. The per-function reference lives on the Standard Library page; what matters here is how it behaves inside a pipeline.

import http from "std/http"
import json from "std/json"

io fn fetchPage(url: string) -> HttpResponse {
  return http.get(url, { accept: "application/json" }, 5000)
}

Incremental runs

A pipeline step that runs on a schedule needs to remember where it stopped. Luma does not invent a watermark abstraction, because a watermark is just a stored string:

import state from "std/state"
import col   from "std/collections"

since  = state.read("orders.since")
rows   = query Orders(since)
newest = col.last(col.sortBy(rows, "created_at"))

// … process rows …

marker = state.write("orders.since", newest.created_at)

parallel over: fan-out across data

parallel fans out over declared branches — a fixed set of things you wrote down. parallel over fans out over data, one body evaluated per element:

classified = parallel over rows as row limit 4 -> Classified[] {
  result = pipeline.classify(row)
  return result
}

limit 1   →   2, 4, 6     history is shared and grows
limit 3   →   2, 2, 2     all three elements race the same store

Flags and ceilings

Everything in a Luma program is a value in memory. There is no streaming and no lazy sequence, so the honest engineering answer is a documented ceiling that fails by name rather than truncating silently or growing until the process dies.

Error: Result set exceeded the row limit (200000). Process in batches with col.chunk, or raise --max-rows.

When a set is genuinely larger than the ceiling, the shape is a batched loop driven by the watermark: query a bounded window, process it, advance the watermark, exit. The next scheduled run picks up where this one stopped. That is the same incremental machinery from the section above, used deliberately rather than as a recovery mechanism.

Transpile: telemetry is tree-shaken from this program (no model or tool calls; DECISIONS D6).

A complete step

Everything above in the shape it actually takes. The orchestrator runs this on a schedule; the program itself has no opinion about when.

import col   from "std/collections"
import state from "std/state"

schema Order {
  id: string
  amount: float
  created_at: string @fmt("date-time")
}

schema Classified {
  id: string
  band: enum["small", "large"]
}

schema Summary {
  processed: int
  advanced_to: string
}

datasource Warehouse {
  credential: "PG_URL"
  pool_max: 4
  timeout: 30000
  retry: 2
}

query Orders(since: string) -> Order[] {
  source: Warehouse
  sql: "select id, amount, created_at from orders where created_at > $1 order by created_at limit 1000"
}

command MarkExported(since: string, batch: string) -> int {
  source: Warehouse
  sql: "update orders set export_batch = $2 where created_at > $1"
  side_effects: "write"
}

agent Banding {
  fn classify(order: Order) -> Classified {
    ask "Band this order by size.\n\nORDER: {order}" -> Classified
  }
}

banding = Banding()

since      = state.read("orders.since")
rows       = query Orders(since)
ordered    = col.sortBy(rows, "created_at")
classified = parallel over ordered as order limit 4 -> Classified[] {
  return banding.classify(order)
}
exported   = command MarkExported(since, "batch-1")
newest     = col.last(ordered)
marker     = state.write("orders.since", newest.created_at)

Read it as five moves: where did I stop · what is new · reshape · the model, bounded · write back and advance. The limit 1000 in the SQL and the limit 4 on the fan-out are the two places this step declares how much of the world it will take on in one run.