KUJO THE COURSESTAGE 5 / AI-native programming · LESSON 31
Runtime contract
ai_stream_chat accepts an optional callback receiving a text delta and raw chunk dictionary. Returning false cancels later chunk delivery; other return values continue. The helper retains an aggregate Result envelope with stream text, chunks, and available metadata.
Replay delivers chunks in recorded order without opening a socket. That lets you test UI or processing behavior without depending on live token timing. The course fixture has a fixed stream, and the callback prints only its first delivered delta before requesting cancellation.
Mental model
Treat streamed output as incomplete until the application has enough evidence to accept a result. A fragment of JSON is not a valid domain object. Keep progressive display separate from the evaluator that decides whether the final structured output is usable.
Cancellation is also a lifecycle event, not automatic success. It can be user-requested, policy-triggered, or caused by an external client disconnect. The application should state whether partial output is retained, discarded, or labeled incomplete. Do not promise that a local cancellation signal reverses work already performed by a remote provider.
Professional pattern
Bound accumulated output and callback work. A callback that performs expensive or unbounded effects can turn a streaming interface into a stalled pipeline. Test the first chunk, complete delivery, and early cancellation independently. Record whether a result was complete before presenting it as evidence.
Common mistakes
Do not equate a visible token with successful work. Do not validate partial structured output as though it were complete. Do not add timing assertions to a replay test unless timing itself is the documented contract. The default exercise tests delivery order and cancellation semantics, not provider latency.
Working example
func first_chunk(delta, raw) { print(delta) return false }
let opts := {"endpoint": "http://127.0.0.1:1/v1/chat/completions", "model": "gpt-replay", "cassette": {"mode": "strict", "dir": "fixtures/ai"}}
match ai_stream_chat("Stream please", opts, first_chunk) {
case Ok(payload): { print("stream callback completed") }
case Err(err): { throw(to_string(err)) }
}
Run it
From the course repository root, use the pinned Kujo 1.3.1 runtime.
kujo check examples/31.kujo
kujo run --untrusted --allow-ai examples/31.kujo
Captured output
chunk one
stream callback completed
Break it and diagnose it
The callback is not callable. The runtime must reject the invalid callback contract.
ai_stream_chat("Stream please", {"endpoint": "http://127.0.0.1:1/v1/chat/completions", "model": "gpt-replay", "cassette": {"mode": "strict", "dir": "fixtures/ai"}}, 42)
kujo run --untrusted --allow-ai examples/31-break.kujo
Exit status: 4. Captured diagnostic:
[KUJOVM001] [vm] Runtime Error: ai_stream_chat() third argument must be a function
--> 0:0
Exercise
Write a replay test that collects all chunks and another that cancels after the first. Compare delivery and aggregate metadata. Label partial output explicitly and prevent it from passing final schema evaluation.
Checkpoint
- I distinguish progressive output from accepted results.
- I can cancel later chunk delivery.
- I test streaming without live timing.
Verified 2026-09-06 · Official Kujo 1.3.1 release · Source contract · Download example