Çalışma Zamanı
Veri & Entegrasyon
Bir Luma programı genellikle daha büyük bir pipeline'ın içindeki tek bir adımdır: yukarıdan bir şey onu zamanlar, program veriyi çeker, yeniden şekillendirir, zor kısmı modele gönderir ve sonucu geri yazar. Bu sayfa o adım hakkında — üst düzey program gövdesinin ne yapabildiği ve onu şekillendiren kurallar.
Üst düzey gövde düz bir sıradır
Bu sayfadaki her şey — query, command, parallel over — bir üst düzey operation. Hiçbiri agent method'unun ya da fn gövdesinin içinde yer alamaz. Parser adıyla reddeder:
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[]
Bunu keşfetmek yerine açıkça söylemeye değer: koşullu olarak, bir döngü içinde ya da bir method'da türettiğin bir değerden sorgu yapamazsın. Programın veri akışı üst düzeyde düz bir sıradır; dallanma, satırlar geldikten sonra onlarla ne yaptığında yaşar.
Agent method'u runtime'ın hakkında akıl yürütebildiği bir birimdir: sahibi olan bir agent'ı, model bağlaması, memory scope'u, retry politikası vardır. Bir query'nin bunların hiçbiri yoktur — o bir agent'a değil, bir datasource'a aittir. parallel over aynı hikâyenin öteki ucu: sahibi olan bir agent'ı hiç yok, runtime sentetik bir tane kuruyor — her elemanın kendi taze dependency instance'ını almasının sebebi de bu.
Bunlara method içinde izin vermek, henüz kimsenin cevaplamadığı soruları cevaplamak demek olurdu — sorgu agent'ın retry'ını devralır mı? memory'sini? token bütçesine sayılır mı? Cevaplar netleşene kadar parse zamanında reddetmek dürüst duruş.
Ortaya çıkan şekil
Bir pipeline adımı bu yüzden yukarıdan aşağı okunur ve model işi ortasında oturur:
watermark'ı oku state.read("orders.since")
│
satırları çek query Orders(since) ← Order'a karşı doğrulanır
│
yeniden şekillendir col.sortBy · col.groupBy · col.chunk
│
model parallel over rows … { agent.classify(row) }
│
geri yaz command MarkExported(since, batch)
│
watermark'ı ilerlet state.write("orders.since", newest)Bir sorguyu saracak olan koşullu mantık iki yerden birine gider: parametre olarak SQL'in içine, ya da satırları alan bir agent method'una. İkisi de zaten genellikle daha iyi tasarım — birincisi filtrelemeyi index'in olduğu yerde tutar, ikincisi kararı test edilebilir tutar.
datasource · query · command
Luma tipli contract'ı sahiplenir, sürücüyü değil. Bir veritabanı sürücüsü npm paketidir ve onu paketlemek transpiler'ın tek dosya, sıfır bağımlılık sözünü kırardı — o yüzden dil sınırı tarif eder, sürücü kodunun yaklaşık yirmi satırını sen verirsin.
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")Bildirim alanları
| Alan | Nerede | Varsayılan | Anlamı |
|---|---|---|---|
credential | datasource | yok | Ortamdan getEnv gibi çözülür. Prompt'a, trace'e ya da cache'e asla girmez. Opsiyonel — başka türlü kimlik doğrulayan bir köprü yazmaz. |
pool_max | datasource | 1 | Dekorasyon değil, gerçek bir eşzamanlılık semaforu. Varsayılan 1'de tek bir datasource'a giden sorgular serileşir — parallel over altında da. Bilinçli olarak yükselt. |
timeout | datasource | zorunlu | Milisaniye. Varsayılanı yok, çünkü sessiz ve sınırsız bekleme makul bir geri düşüş değil. |
retry | datasource | 0 | Yalnızca okumalar için. Bir command asla yeniden denenmez — idempotent olmayabilecek bir yazma, dilin arkandan tekrarlayacağı bir şey değil. |
source | query · command | zorunlu | Bu ifadenin hangi datasource'a ait olduğu. |
sql | query · command | zorunlu | Tek satırlık string. Parametreler sıralı $1…$n, bildirilen parametrelerle sırayla eşlenir. |
side_effects: "write" | command | zorunlu | command bunu bildirmek zorunda; query bildiremez. Bildirim biçimi ile etki birbiriyle çelişemez. |
Köprü sende
Bir tool modülü bağlantıları bilinen bir ad altında dışa açar. Sınırın Luma'nın sahiplenmediği tek parçası bu:
export const datasources = {
Warehouse: async (sql, params, signal) => {
const result = await pool.query(sql, params); // pg, mysql2, better-sqlite3 — senin seçimin
return result.rows;
},
};Köprü yoksa program hiçbir şey çalıştırmadan durur: Error: datasource 'Warehouse' has no registered bridge; export it from a tool module as datasources.Warehouse.
Satırlar bildirilen şemaya karşı doğrulanır. Yukarıda tipi değişmiş bir kolon, dönüşümün ortasında ya da daha kötüsü sessizce bir model prompt'unun içinde değil, sınırda — satır indeksi ve alan adıyla — düşer: NewOrders.row[0].amount must be a number.
sql interpolation yapamaz. İçine {since} yazmak parse sırasında reddedilir: Error: query 'Orders' sql cannot interpolate; use $1 parameters. Injection yüzeyi kod incelemesinde değil, dilde kapanır.
Bir satır, ya da çok
Schema[] dönüş tipi ne gelirse alır. Çıplak Schema ise tam olarak bir satır iddia eder ve aksi hâlde düşer — bir agregat için istediğin tam olarak budur:
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.
Bir command tek seferde tek ifade çalıştırır ve dil transaction bildirimi sunmaz. Yazmaya kalkışmak yok sayılmaz, adıyla reddedilir:
Error: Datasource 'Warehouse' does not support 'transaction'; it is deliberately out of this package — handle transactions inside the bridge (ROADMAP § U4).
Çok ifadeli atomiklik, sürücünün zaten sahip olduğu yere — köprüye — aittir: bağlantıyı al, transaction'ı aç, gerekeni koştur, commit et. Bunu Luma'da açmak bağlantı ömrünü dilde modellemek demek olurdu ve veri yüzeyi bunu bilerek yapmıyor.
Pratik sonucu: bir adım ya hep ya hiç olmak zorundaysa onu SQL'i tek atomik ifade olan tek bir command olarak yaz, ya da bütün birimi köprünün içine taşı.
HTTP
std/http veritabanı olmayan servislere ulaşır — bir API, bir webhook, HTTP arayüzü olan bir object store. Fonksiyon başına referans Standard Library sayfasında; burada önemli olan pipeline içinde nasıl davrandığı.
import http from "std/http"
import json from "std/json"
io fn fetchPage(url: string) -> HttpResponse {
return http.get(url, { accept: "application/json" }, 5000)
}| Davranış | Neden böyle |
|---|---|
| Statü kodu hata değil | 404 ve 500 statüyle birlikte normal döner. Bir 404'ün "bu kaydı atla" mı yoksa "koşuyu durdur" mu demek olduğuna yalnızca program karar verebilir. Runtime'ın yükselttiği tek başarısızlık host'a hiç ulaşamamak. |
timeout zorunlu argüman | Varsayılanı yok, çünkü bir batch işin içindeki sessiz ve sınırsız bekleme, pipeline'ı gece üçte düşüren başarısızlık kipidir. |
Retry http'nin işi değil | Agent'ın retry'sini ya da çağrının etrafında bir try kullan. İstemcinin içine ikinci bir retry koymak, etkin deneme sayısını kimsenin takip etmediği iki sayının çarpımı yapardı. |
| Gövde sınırlı | --max-response-bytes, varsayılan 10 MB, aşılınca adıyla reddedilir. |
Maliyet. Expression evaluator senkron ve senkron bir fetch yok, o yüzden her çağrı bir alt süreçte koşuyor. Kaba taneli iş için sorun değil — sayfalanmış bir çekme, birkaç yüz istek — ama büyük bir kümenin her satırını zenginleştirmek için yanlış araç; orada süreç maliyeti koşuya baskın çıkar. On bin satır başına çağrı, beş ile on üç dakika saf ek yük demek. § U5 olarak izleniyor.
Redaksiyon. İstek header'ları trace'e iki filtreden geçerek ulaşır ve kural header'ın adıyla ilgilidir, hassaslığıyla değil. Adı api_key · apikey · api-key · password · secret · credential · authorization ile eşleşen (büyük/küçük harf duyarsız, alt dize) bir header'ın değerinin tamamı [REDACTED] ile değiştirilir; ayrıca runtime'ın zaten bildiği her sır değeri — ortamdan ve model profillerinden çözülen api key'ler — geçtiği her yerde değiştirilir. Diğer her header düz metin yazılır: cookie, x-auth-token, x-session, bir imza header'ı. x-api-key'in kapsandığına dikkat et — adı eşleştiği için. Sırrı, kalıbın adlandırmadığı bir header'da taşırsan trace'e düşer.
Artımlı koşular
Zamanlanmış koşan bir pipeline adımının nerede durduğunu hatırlaması gerekir. Luma bir watermark soyutlaması icat etmez, çünkü watermark yalnızca saklanmış bir string'dir:
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"))
// … satırlar işlenir …
marker = state.write("orders.since", newest.created_at)| Kural | Ayrıntı |
|---|---|
Dosyayı --state-file adlandırır | Varsayılan konum yok. Bayraksız bir koşu, senin seçmediğin bir yere yazmak yerine adıyla reddeder — sessiz varsayılan, iki iş aynı makineyi paylaştığı ilk anda sessiz veri kaybıdır. |
| Yazma atomik | Geçici dosyaya yazılır ve rename edilir. Yazmanın ortasında öldürülen bir koşu, kırpılmış bir dosya değil, önceki state'i bırakır. |
Değerler daima string | Sayı ya da zaman damgası saklamak, onu json ya da date ile kendin dönüştürmek demek. Depo bir tip sistemi icat etmez. |
--record/--replay'in parçası değil | Replay model çağrılarını kapsar. State diskteki gerçek bir dosyadır ve replay edilen koşu onu gerçekten okur ve yazar — iki çalıştırmayı karşılaştırırken koşu başına izole et. |
Watermark'ı, tarif ettiği iş commit edildikten sonra yaz, önce değil. Koşu ortada ölürse bir sonraki koşu bir partiyi tekrar işler — doğru yazılmış bir pipeline buna tahammül eder — bir partiyi atlamak yerine, ki ona hiçbir şey tahammül etmez.
parallel over: veri üzerinde fan-out
parallel bildirilen dallar üzerinde fan-out yapar — yazdığın sabit bir küme. parallel over veri üzerinde fan-out yapar, eleman başına bir gövde değerlendirilir:
classified = parallel over rows as row limit 4 -> Classified[] {
result = pipeline.classify(row)
return result
}| Kural | Ayrıntı |
|---|---|
| Sonuçlar girdi sırasında | Tamamlanma sırasında değil. Tasarımın pazarlık etmeyeceği garanti bu: interpreter ile transpiled artifact byte-byte karşılaştırılıyor ve nondeterministik bir sıra o karşılaştırmayı değersiz kılardı. Yavaş bir eleman sona kaymaz. |
limit zorunlu | Varsayılan eşzamanlılık yok. Bir sorgu sonucu üzerinde sınırsız fan-out, kendi veritabanına ya da provider'ına karşı kazara bir hizmet reddidir. |
Dönüş tipi Schema[] olmalı | Gövde eleman başına bir Schema üretir ve her eleman sonucu tek tek doğrulanır; böylece bozuk tek bir eleman partiyi bozmak yerine adıyla anılır. |
| Seçenek sırası sabit | on_error, limit'ten önce gelir: … as row on_error continue limit 4 -> Out[]. Ters yazılınca header hiç tanınmaz, gövde satırları üst düzeye düşer ve aldığın şey 'return' can only be used in an agent method or fn body (line N). olur — parallel over'dan hiç söz etmeyen bir mesaj. |
Hata varsayılanı fail_fast | on_error continue düşen elemanı atar — sonuç listesi delik taşımak yerine kapanır — ve atlanan sayısı parallel_over tipinde, { total, skipped } metadata'sıyla bir telemetry event'i olarak yayılır. O event yalnızca bir şey atlandığında çıkar, yani yokluğu hiçbir şeyin kaybolmadığı anlamına gelir — kısalmış bir listenin makine tarafından okunabilir tek sinyali odur (Telemetry & Trace'e bak). on_error fallback adıyla reddedilir: eleman başına bir fallback değeri tanımlı değil ve uydurmak sessizce veri ikame etmek olurdu. |
| İç içe kullanım yok | parallel over içinde parallel over parse sırasında reddedilir. İç içe eşzamanlılık çarpar ve limit'in tarif edebileceği bir şey olmaktan çıkar. |
| Eleman başına taze instance | Gövdede anılan global agent instance'ları eleman başına bir instance key altında çözülür. Varsayılan memory_scope: instance altında bu, her elemana kendi geçmişini verir. Diğer scope'lar için geçerli değildir — aşağıdaki uyarıya bak. |
Süslü parantezlerin içinde bir method gövdesinin izin verdiğini yazabilirsin — ask, tool çağrıları, match, for, bir dependency çağrısı. Ama parallel over'ın kendisi üst düzey bir operation'dır ve bir method'un içine yazılamaz. Agent sayfasında değil burada belgelenmesinin sebebi tam olarak bu.
Eleman izolasyonu instance key'den gelir, ama agent geçmişi scope'a göre anahtarlanır. Yalnızca varsayılan instance scope'u instance başına anahtarlanır; session, shared ve persistent agent adına göre anahtarlanır, yani bir parallel over'ın her elemanı tek bir geçmişi okur ve yazar.
Mock gateway'e karşı, memory_scope: "session" olan bir agent ile üç satır üzerinde, giden her istekteki history mesaj sayısı sayılarak ölçüldü:
limit 1 → 2, 4, 6 geçmiş paylaşılıyor ve büyüyor limit 3 → 2, 2, 2 üç eleman da aynı depoyu yarışıyor
Yani modele gerçekte gönderilen prompt limit'e bağlı hâle geliyor — ki o başka türlü yalnızca bir eşzamanlılık tavanı — ve aynı limit'teki iki koşu, zamanlamaya göre farklı prompt gönderebiliyor. parallel over içinde agent'ları varsayılan memory scope'unda tut. Elemanların gerçekten paylaşılan state'e ihtiyacı varsa onu fan-out'tan sonra std/state'e koy; orada yazma sırası senin kontrolünde.
Bu ikisi etkileşir ve etkileşim insanları şaşırtır. pool_max'ı varsayılan 1 olan bir datasource ile limit 8, tek bir bağlantı için yarışan sekiz eşzamanlı gövde demektir: model çağrıları örtüşür, sorgular örtüşmez. Gövdeler sorgu yapıyorsa pool_max'ı gerçekten istediğin eşzamanlılığa yükselt.
Bayraklar ve tavanlar
Bir Luma programındaki her şey bellekteki bir değerdir. Streaming ve lazy sequence yok, dolayısıyla dürüst mühendislik cevabı: sessizce kırpmak ya da süreç ölene kadar büyümek yerine adıyla düşen belgelenmiş bir tavan.
| Bayrak | Varsayılan | Neyi sınırlar | Neyi sınırlamaz |
|---|---|---|---|
--max-rows | 200000 | Bir query'nin döndürdüğü satırlar ve io.readFiles'ın okuduğu dosyalar | state.keys() ve bellekte kendin kurduğun her şey |
--max-response-bytes | 10 MB | Tek bir std/http yanıt gövdesi | Bir koşudaki toplam |
--state-file | yok | std/state'in okuyup yazdığı yer | — state.* hiç çağrıldığında zorunlu, state.keys() dahil |
Error: Result set exceeded the row limit (200000). Process in batches with col.chunk, or raise --max-rows.
Bir küme gerçekten tavandan büyük olduğunda şekil, watermark'ın sürdüğü partili bir döngüdür: sınırlı bir pencere sorgula, işle, watermark'ı ilerlet, çık. Bir sonraki zamanlanmış koşu bunun bıraktığı yerden devam eder. Bu, yukarıdaki bölümdeki aynı artımlı makine — kurtarma mekanizması olarak değil, bilinçli olarak kullanılmış hâli.
--state-file, --max-rows ve --max-response-bytes transpiled bir programda da çalışır. Veri-yalnız bir programın kaybettiği şey telemetri. Artifact trace makinesini yalnızca programda bir model çağrısı ya da bir tool çağrısı varsa taşır; ikisi de yoksa tree-shake edilir ve --trace, --trace-file, --trace-jsonl, --otel-file hepsi adıyla reddedilir:
Transpile: telemetry is tree-shaken from this program (no model or tool calls; DECISIONS D6).
Aynı programı koşturan interpreter trace yazar ve o trace query ile command event'lerini içerir. İki yol burada bilerek ayrışıyor — adıyla reddetmek, interpreter'ın yazacağından farklı bir trace'i sessizce yazan bir artifact'tan iyidir. Tek dosya olarak dağıtmak'a bak.
Tam bir adım
Yukarıdakilerin hepsi, gerçekte aldığı şekliyle. Orchestrator bunu zamanlanmış koşturur; programın kendisinin ne zaman koşacağı hakkında bir görüşü yoktur.
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)Beş hamle olarak oku: nerede durmuştum · yeni olan ne · yeniden şekillendir · model, sınırlı · geri yaz ve ilerlet. SQL'deki limit 1000 ile fan-out'taki limit 4, bu adımın tek koşuda dünyanın ne kadarını üstleneceğini bildirdiği iki yer.
Scheduling, DAG bağımlılıkları, backfill ve adımın tamamının yeniden denenmesi Luma'nın işi değil ve olmayacak. Bu program tek bir düğüm; grafiği senin orchestrator'ın sahipleniyor. Dilin diğer sözlerini tutabilecek kadar küçük kalmasının sebebi bu sınır — parse zamanı garantileri, tek dosyalık artifact, iki çalıştırma yolu arasındaki byte-byte eşitlik.