diff --git a/CHANGELOG.md b/CHANGELOG.md
index 3186215ba..c608c9e20 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,15 @@
 
 For the latest version, see [changelogs/v0.18-current.md](changelogs/v0.18-current.md).
 
+## v0.32.0 (Unreleased)
+
+- Added experimental `std/ai.stepWithStreamRecorded`, originally authored by
+  [@arniwesth](https://github.com/arniwesth). It preserves immediate stream
+  callbacks while returning the exact ordered adapter-emitted chunk log and
+  typed terminal outcome, including an explicit incomplete prefix on an
+  unencodable chunk. See [#546](https://github.com/sunholo-data/ailang/issues/546)
+  and [arniwesth/ailang#2](https://github.com/arniwesth/ailang/pull/2).
+
 ## Changelog Archives
 
 The full changelog has been split into themed files for searchability and readability:
diff --git a/internal/builtins/ai.go b/internal/builtins/ai.go
index a2dce829a..1edf2cc6e 100644
--- a/internal/builtins/ai.go
+++ b/internal/builtins/ai.go
@@ -24,6 +24,7 @@ func init() {
 	registerAIStepWithCache()
 	// M-AI-STEP-STREAMING (v0.18.7) — typed-StepResult streaming variant
 	registerAIStepWithStream()
+	registerAIStepWithStreamRecorded()
 }
 
 // _ai_call: Call the AI oracle with a string input
diff --git a/internal/builtins/ai_step.go b/internal/builtins/ai_step.go
index 81bfbb1b3..7cc640cbb 100644
--- a/internal/builtins/ai_step.go
+++ b/internal/builtins/ai_step.go
@@ -451,6 +451,87 @@ func streamUsageRecordType(T *types.Builder) types.Type {
 	)
 }
 
+// ============================================================================
+// _ai_step_with_stream_recorded — PROTOTYPE of the proposed upstream
+// recorded-stream API. Preserves immediate callbacks and additionally returns
+// the exact ordered observed chunks with the final outcome, on BOTH outcomes.
+// NOT an upstream feature; see the motoko 009 spike.
+// ============================================================================
+
+func makeAIStepWithStreamRecordedType() types.Type {
+	T := types.NewBuilder()
+	onChunkType := T.Func(streamChunkType(T)).Returns(T.Unit()).Build()
+	return T.Func(
+		T.String(),
+		T.List(messageRecordType(T)),
+		T.List(toolSchemaRecordType(T)),
+		T.List(cacheBreakpointRecordType(T)),
+		onChunkType,
+	).
+		Returns(T.Record(
+			types.Field("chunks", T.List(streamChunkType(T))),
+			types.Field("outcome", T.App("Result", stepResultRecordType(T), aiErrorRecordType(T))),
+		)).
+		Effects("AI")
+}
+
+func aiStepWithStreamRecordedImpl(ctx *effects.EffContext, args []eval.Value) (eval.Value, error) {
+	if err := ctx.RequireCapWithBudget("AI", ""); err != nil {
+		return nil, err
+	}
+	return effects.Call(ctx, "AI", "stepWithStreamRecorded", args)
+}
+
+func registerAIStepWithStreamRecorded() {
+	err := RegisterEffectBuiltin(BuiltinSpec{
+		Module:  "std/ai",
+		Name:    "_ai_step_with_stream_recorded",
+		NumArgs: 5,
+		Effect:  "AI",
+		Type:    makeAIStepWithStreamRecordedType,
+		Impl:    aiStepWithStreamRecordedImpl,
+		Metadata: &BuiltinMetadata{
+			Description: "Streaming multi-turn AI completion that preserves immediate per-chunk callbacks and returns the exact ordered observed chunks with the final outcome",
+			LongDesc: `Identical to _ai_step_with_stream except the return shape:
+
+  { chunks: [StreamChunk], outcome: Result[StepResult, AIError] }
+
+The chunks are returned on BOTH outcomes, so a stream that fails part-way
+still yields every chunk observed before the failure. Callbacks still fire
+immediately at arrival; the returned log is an exact, non-duplicating record
+of the chunks the provider adapter emits, not of the provider wire. In
+particular, tool-call input_json stream content is not emitted as chunks.
+
+The returned log is retained linearly and without a bound in memory until the
+call returns. If a chunk cannot be encoded, the outcome is a non-retryable
+Internal error whose stable message prefix is "unencodable stream chunk"; the
+returned chunks are explicitly an incomplete prefix. From that point AILANG
+records, encodes, and delivers nothing, and its per-chunk drain work is bounded,
+but the call still returns only when the provider's stream ends.
+
+This fail-loud behavior deliberately differs from _ai_step_with_stream, which
+silently skips an unencodable chunk. The siblings are equivalent only for
+fully encodable streams, which includes every stream constructible today.`,
+			Params: []ParamDoc{
+				{Name: "model", Description: "Model ID (or empty for handler default)"},
+				{Name: "messages", Description: "Conversation as list[Message]"},
+				{Name: "tools", Description: "Tool catalog as list[ToolSchema]"},
+				{Name: "cache_breakpoints", Description: "Opt-in cache hints as list[CacheBreakpoint]"},
+				{Name: "on_chunk", Description: "Callback (StreamChunk) -> () invoked per chunk"},
+			},
+			Returns:   "{ chunks: [StreamChunk], outcome: Result[StepResult, AIError] }",
+			SeeAlso:   []string{"_ai_step_with_stream", "std/ai.stepWithStreamRecorded"},
+			Since:     "v0.32.0",
+			Stability: StabilityExperimental,
+			Tags:      []string{"ai", "result", "streaming", "recorded"},
+			Category:  "ai",
+		},
+	})
+	if err != nil {
+		panic("failed to register _ai_step_with_stream_recorded builtin: " + err.Error())
+	}
+}
+
 func registerAIStepWithStream() {
 	err := RegisterEffectBuiltin(BuiltinSpec{
 		Module:  "std/ai",
diff --git a/internal/effects/ai_decode.go b/internal/effects/ai_decode.go
new file mode 100644
index 000000000..43d2c054e
--- /dev/null
+++ b/internal/effects/ai_decode.go
@@ -0,0 +1,148 @@
+package effects
+
+import (
+	"fmt"
+
+	"github.com/sunholo-data/ailang/internal/ai"
+	"github.com/sunholo-data/ailang/internal/eval"
+)
+
+// ============================================================================
+// Decoders — AILANG records → Go structs
+// ============================================================================
+
+// decodeMessages converts an AILANG list of Message records into []ai.Message.
+// Tolerates missing fields (treated as zero values) so callers can omit
+// optional fields like ToolCalls / ToolCallID. Rejects non-record entries.
+func decodeMessages(list *eval.ListValue) ([]ai.Message, error) {
+	out := make([]ai.Message, 0, len(list.Elements))
+	for i, elem := range list.Elements {
+		rec, ok := elem.(*eval.RecordValue)
+		if !ok {
+			return nil, fmt.Errorf("messages[%d]: expected record, got %T", i, elem)
+		}
+		msg := ai.Message{
+			Role:       getStringField(rec, "role"),
+			Content:    getStringField(rec, "content"),
+			ToolCallID: getStringField(rec, "tool_call_id"),
+		}
+		// tool_calls is an optional list[ToolCall]; absent or nil = no calls.
+		if tcList, ok := rec.Fields["tool_calls"].(*eval.ListValue); ok {
+			calls, err := decodeToolCalls(tcList)
+			if err != nil {
+				return nil, fmt.Errorf("messages[%d]: %w", i, err)
+			}
+			msg.ToolCalls = calls
+		}
+		// images is a list[ImagePart] (M-STD-AI-VISION-INPUT). Empty/absent =
+		// text-only message, wire-identical to pre-vision (Images stays nil).
+		if imgList, ok := rec.Fields["images"].(*eval.ListValue); ok && len(imgList.Elements) > 0 {
+			images, err := decodeImageParts(imgList)
+			if err != nil {
+				return nil, fmt.Errorf("messages[%d]: %w", i, err)
+			}
+			msg.Images = images
+		}
+		out = append(out, msg)
+	}
+	return out, nil
+}
+
+// decodeToolCalls converts an AILANG list of ToolCall records into []ai.ToolCall.
+func decodeToolCalls(list *eval.ListValue) ([]ai.ToolCall, error) {
+	out := make([]ai.ToolCall, 0, len(list.Elements))
+	for i, elem := range list.Elements {
+		rec, ok := elem.(*eval.RecordValue)
+		if !ok {
+			return nil, fmt.Errorf("tool_calls[%d]: expected record, got %T", i, elem)
+		}
+		out = append(out, ai.ToolCall{
+			ID:        getStringField(rec, "id"),
+			Name:      getStringField(rec, "name"),
+			Arguments: getStringField(rec, "arguments"),
+		})
+	}
+	return out, nil
+}
+
+// decodeImageParts decodes an AILANG list[ImagePart] into []ai.ImagePart.
+// Each element is a {source, mime} record (M-STD-AI-VISION-INPUT). An empty
+// source is rejected — the caller (decodeMessages) surfaces it as a typed
+// SchemaValidation AIError rather than silently forwarding a blank image.
+func decodeImageParts(list *eval.ListValue) ([]ai.ImagePart, error) {
+	out := make([]ai.ImagePart, 0, len(list.Elements))
+	for i, elem := range list.Elements {
+		rec, ok := elem.(*eval.RecordValue)
+		if !ok {
+			return nil, fmt.Errorf("images[%d]: expected record, got %T", i, elem)
+		}
+		source := getStringField(rec, "source")
+		if source == "" {
+			return nil, fmt.Errorf("images[%d]: empty source (expected base64 or data-URI)", i)
+		}
+		out = append(out, ai.ImagePart{
+			Source: source,
+			Mime:   getStringField(rec, "mime"),
+		})
+	}
+	return out, nil
+}
+
+// decodeCacheBreakpoints converts an AILANG list of CacheBreakpoint records
+// into []ai.CacheBreakpoint. Tolerates missing fields (treated as empty
+// strings) — provider-side validation rejects unknown positions with a
+// once-per-session warning rather than failing.
+func decodeCacheBreakpoints(list *eval.ListValue) ([]ai.CacheBreakpoint, error) {
+	if list == nil || len(list.Elements) == 0 {
+		return nil, nil
+	}
+	out := make([]ai.CacheBreakpoint, 0, len(list.Elements))
+	for i, elem := range list.Elements {
+		rec, ok := elem.(*eval.RecordValue)
+		if !ok {
+			return nil, fmt.Errorf("cache_breakpoints[%d]: expected record, got %T", i, elem)
+		}
+		out = append(out, ai.CacheBreakpoint{
+			Position: getStringField(rec, "position"),
+			TTL:      getStringField(rec, "ttl"),
+		})
+	}
+	return out, nil
+}
+
+// decodeToolSchemas converts an AILANG list of ToolSchema records into
+// []ai.ToolSchema.
+func decodeToolSchemas(list *eval.ListValue) ([]ai.ToolSchema, error) {
+	out := make([]ai.ToolSchema, 0, len(list.Elements))
+	for i, elem := range list.Elements {
+		rec, ok := elem.(*eval.RecordValue)
+		if !ok {
+			return nil, fmt.Errorf("tools[%d]: expected record, got %T", i, elem)
+		}
+		out = append(out, ai.ToolSchema{
+			Name:        getStringField(rec, "name"),
+			Description: getStringField(rec, "description"),
+			Parameters:  getStringField(rec, "parameters"),
+		})
+	}
+	return out, nil
+}
+
+// getStringField extracts a string field from a record, returning ""
+// if the field is missing or not a StringValue. Forgiving by design —
+// schema-level rigor is the type checker's job, not this layer.
+func getStringField(rec *eval.RecordValue, name string) string {
+	v, ok := rec.Fields[name]
+	if !ok {
+		return ""
+	}
+	s, ok := v.(*eval.StringValue)
+	if !ok {
+		return ""
+	}
+	return s.Value
+}
+
+// ============================================================================
+// Encoders — Go structs → AILANG Result records
+// ============================================================================
diff --git a/internal/effects/ai_encode.go b/internal/effects/ai_encode.go
new file mode 100644
index 000000000..73d58d59f
--- /dev/null
+++ b/internal/effects/ai_encode.go
@@ -0,0 +1,120 @@
+package effects
+
+import (
+	"github.com/sunholo-data/ailang/internal/ai"
+	"github.com/sunholo-data/ailang/internal/eval"
+)
+
+// encodeStreamChunk converts a Go ai.StreamChunk variant into the matching
+// AILANG `StreamChunk` ADT. Mirrors the type definitions in std/ai.ail
+// (see M-AI-STEP-STREAMING design doc for shape contract).
+//
+//	ai.StreamContentDelta{Text}  → ContentDelta(string)
+//	ai.StreamThinkingDelta{Text} → ThinkingDelta(string)  (v0.18.8)
+//	ai.StreamUsage{...}          → Usage({input_tokens, output_tokens,
+//	                                     cache_read_input_tokens,
+//	                                     cache_creation_input_tokens})
+func encodeStreamChunk(chunk ai.StreamChunk) eval.Value {
+	switch c := chunk.(type) {
+	case ai.StreamContentDelta:
+		return &eval.TaggedValue{
+			CtorName: "ContentDelta",
+			Fields:   []eval.Value{&eval.StringValue{Value: c.Text}},
+		}
+	case ai.StreamThinkingDelta:
+		return &eval.TaggedValue{
+			CtorName: "ThinkingDelta",
+			Fields:   []eval.Value{&eval.StringValue{Value: c.Text}},
+		}
+	case ai.StreamUsage:
+		usageRec := &eval.RecordValue{
+			Fields: map[string]eval.Value{
+				"input_tokens":                &eval.IntValue{Value: c.InputTokens},
+				"output_tokens":               &eval.IntValue{Value: c.OutputTokens},
+				"cache_read_input_tokens":     &eval.IntValue{Value: c.CacheReadInputTokens},
+				"cache_creation_input_tokens": &eval.IntValue{Value: c.CacheCreationInputTokens},
+			},
+		}
+		return &eval.TaggedValue{
+			CtorName: "Usage",
+			Fields:   []eval.Value{usageRec},
+		}
+	default:
+		return nil
+	}
+}
+
+// makeOkStringResult builds Ok(string) — for callResult / callJsonResult.
+func makeOkStringResult(s string) eval.Value {
+	return &eval.TaggedValue{
+		CtorName: "Ok",
+		Fields:   []eval.Value{&eval.StringValue{Value: s}},
+	}
+}
+
+// makeOkStepResult builds Ok(StepResult record) — for step.
+func makeOkStepResult(resp *ai.Response) eval.Value {
+	// Build the assistant Message record.
+	msgRec := &eval.RecordValue{
+		Fields: map[string]eval.Value{
+			"role":         &eval.StringValue{Value: "assistant"},
+			"content":      &eval.StringValue{Value: resp.Text},
+			"tool_calls":   encodeToolCalls(resp.ToolCalls),
+			"tool_call_id": &eval.StringValue{Value: ""},
+		},
+	}
+	stepResult := &eval.RecordValue{
+		Fields: map[string]eval.Value{
+			"message":                     msgRec,
+			"tool_calls":                  encodeToolCalls(resp.ToolCalls),
+			"finish_reason":               &eval.StringValue{Value: resp.FinishReason},
+			"input_tokens":                &eval.IntValue{Value: resp.InputTokens},
+			"output_tokens":               &eval.IntValue{Value: resp.OutputTokens},
+			"cache_read_input_tokens":     &eval.IntValue{Value: resp.CacheReadInputTokens},
+			"cache_creation_input_tokens": &eval.IntValue{Value: resp.CacheCreationInputTokens},
+		},
+	}
+	return &eval.TaggedValue{
+		CtorName: "Ok",
+		Fields:   []eval.Value{stepResult},
+	}
+}
+
+// encodeToolCalls builds an AILANG list[ToolCall] from a Go slice.
+func encodeToolCalls(calls []ai.ToolCall) eval.Value {
+	elems := make([]eval.Value, 0, len(calls))
+	for _, c := range calls {
+		elems = append(elems, &eval.RecordValue{
+			Fields: map[string]eval.Value{
+				"id":        &eval.StringValue{Value: c.ID},
+				"name":      &eval.StringValue{Value: c.Name},
+				"arguments": &eval.StringValue{Value: c.Arguments},
+			},
+		})
+	}
+	return &eval.ListValue{Elements: elems}
+}
+
+// makeAIErrorResultRecord builds Err(AIError record) for any AI op that
+// surfaces a typed failure. AIError shape: {code, message, retryable},
+// matching std/ai/streaming.AIError byte-for-byte.
+func makeAIErrorResultRecord(e *ai.AIError) eval.Value {
+	if e == nil {
+		// Defensive: should never happen, but if it does emit an
+		// internal-coded record so downstream consumers see something
+		// meaningful instead of a nil-deref.
+		e = ai.NewAIError(ai.CodeInternal, "nil AIError surfaced from effect op", false)
+	}
+	return &eval.TaggedValue{
+		CtorName: "Err",
+		Fields: []eval.Value{
+			&eval.RecordValue{
+				Fields: map[string]eval.Value{
+					"code":      &eval.StringValue{Value: e.Code},
+					"message":   &eval.StringValue{Value: e.Message},
+					"retryable": &eval.BoolValue{Value: e.Retryable},
+				},
+			},
+		},
+	}
+}
diff --git a/internal/effects/ai_step.go b/internal/effects/ai_step.go
index 37ba18777..410067416 100644
--- a/internal/effects/ai_step.go
+++ b/internal/effects/ai_step.go
@@ -52,6 +52,7 @@ func init() {
 	RegisterOp("AI", "step", aiStep)
 	RegisterOp("AI", "stepWithCache", aiStepWithCache)
 	RegisterOp("AI", "stepWithStream", aiStepWithStream)
+	RegisterOp("AI", "stepWithStreamRecorded", aiStepWithStreamRecorded)
 }
 
 // ============================================================================
@@ -327,339 +328,52 @@ func aiStepWithCache(ctx *EffContext, args []eval.Value) (eval.Value, error) {
 // without native streaming (Gemini, Ollama, configdriven) NO-OP fall back to
 // StepWithCache and fire one synthetic ContentDelta + Usage at the end.
 func aiStepWithStream(ctx *EffContext, args []eval.Value) (eval.Value, error) {
-	if len(args) < 5 {
-		return nil, fmt.Errorf("E_AI_TYPE_ERROR: stepWithStream: expected 5 arguments, got %d", len(args))
-	}
-	model, ok := args[0].(*eval.StringValue)
-	if !ok {
-		return nil, fmt.Errorf("E_AI_TYPE_ERROR: stepWithStream: expected string model, got %T", args[0])
-	}
-	messagesArg, ok := args[1].(*eval.ListValue)
-	if !ok {
-		return nil, fmt.Errorf("E_AI_TYPE_ERROR: stepWithStream: expected list[Message] messages, got %T", args[1])
-	}
-	toolsArg, ok := args[2].(*eval.ListValue)
-	if !ok {
-		return nil, fmt.Errorf("E_AI_TYPE_ERROR: stepWithStream: expected list[ToolSchema] tools, got %T", args[2])
-	}
-	breakpointsArg, ok := args[3].(*eval.ListValue)
-	if !ok {
-		return nil, fmt.Errorf("E_AI_TYPE_ERROR: stepWithStream: expected list[CacheBreakpoint] cache_breakpoints, got %T", args[3])
-	}
-	onChunkFn := args[4]
-	if onChunkFn == nil {
-		return nil, fmt.Errorf("E_AI_TYPE_ERROR: stepWithStream: on_chunk callback is nil")
-	}
-	if ctx.AI == nil {
-		return makeAIErrorResultRecord(ai.NewAIError(ai.CodeProviderNotFound, ErrNoAIHandler.Error(), false)), nil
-	}
-	if ctx.FnCaller == nil {
-		return makeAIErrorResultRecord(ai.NewAIError(ai.CodeInternal, "stepWithStream: FnCaller not wired (evaluator integration missing)", false)), nil
-	}
-
-	messages, conversionErr := decodeMessages(messagesArg)
-	if conversionErr != nil {
-		return makeAIErrorResultRecord(ai.NewAIError(ai.CodeSchemaValidation, conversionErr.Error(), false)), nil
-	}
-	tools, conversionErr := decodeToolSchemas(toolsArg)
-	if conversionErr != nil {
-		return makeAIErrorResultRecord(ai.NewAIError(ai.CodeSchemaValidation, conversionErr.Error(), false)), nil
-	}
-	breakpoints, conversionErr := decodeCacheBreakpoints(breakpointsArg)
-	if conversionErr != nil {
-		return makeAIErrorResultRecord(ai.NewAIError(ai.CodeSchemaValidation, conversionErr.Error(), false)), nil
-	}
-
-	// Wrap the AILANG closure in a Go callback. Errors from the AILANG
-	// callback are logged via the trace channel but DO NOT abort the SSE
-	// drain — the caller still gets a complete StepResult on success.
-	chunkCount := 0
-	onChunk := func(chunk ai.StreamChunk) {
-		chunkCount++
-		encoded := encodeStreamChunk(chunk)
-		if encoded == nil {
-			return
-		}
-		if _, err := ctx.FnCaller(onChunkFn, encoded); err != nil {
-			// Surface as a trace event so dashboards see callback failures
-			// without aborting the stream.
-			ctx.RecordAIEffect("stepWithStream.callback",
-				[]string{fmt.Sprintf("chunk:%d", chunkCount)},
-				fmt.Sprintf(errResultPrefix, err.Error()),
-				nil,
-			)
-		}
-	}
-
-	resp, err := ctx.AI.StepWithStream(model.Value, messages, tools, breakpoints, onChunk)
+	_, resp, aiErr, err := aiStreamCore(ctx, args, "stepWithStream", streamRecordPolicy{})
 	if err != nil {
-		aiErr := classifyOpError(err)
-		ctx.RecordAIEffect("stepWithStream",
-			[]string{truncateForTrace(model.Value), fmt.Sprintf("messages:%d tools:%d cache:%d chunks:%d", len(messages), len(tools), len(breakpoints), chunkCount)},
-			fmt.Sprintf(errResultPrefix, aiErr.Code),
-			ctx.AI.LastRoutingMetadata(),
-		)
+		return nil, err
+	}
+	if aiErr != nil {
 		return makeAIErrorResultRecord(aiErr), nil
 	}
-
-	ctx.RecordAIEffect("stepWithStream",
-		[]string{truncateForTrace(model.Value), fmt.Sprintf("messages:%d tools:%d cache:%d chunks:%d", len(messages), len(tools), len(breakpoints), chunkCount)},
-		fmt.Sprintf("text:%s tool_calls:%d finish:%s cache_read:%d cache_create:%d", truncateForTrace(resp.Text), len(resp.ToolCalls), resp.FinishReason, resp.CacheReadInputTokens, resp.CacheCreationInputTokens),
-		ctx.AI.LastRoutingMetadata(),
-	)
 	return makeOkStepResult(resp), nil
 }
 
-// encodeStreamChunk converts a Go ai.StreamChunk variant into the matching
-// AILANG `StreamChunk` ADT. Mirrors the type definitions in std/ai.ail
-// (see M-AI-STEP-STREAMING design doc for shape contract).
-//
-//	ai.StreamContentDelta{Text}  → ContentDelta(string)
-//	ai.StreamThinkingDelta{Text} → ThinkingDelta(string)  (v0.18.8)
-//	ai.StreamUsage{...}          → Usage({input_tokens, output_tokens,
-//	                                     cache_read_input_tokens,
-//	                                     cache_creation_input_tokens})
-func encodeStreamChunk(chunk ai.StreamChunk) eval.Value {
-	switch c := chunk.(type) {
-	case ai.StreamContentDelta:
-		return &eval.TaggedValue{
-			CtorName: "ContentDelta",
-			Fields:   []eval.Value{&eval.StringValue{Value: c.Text}},
-		}
-	case ai.StreamThinkingDelta:
-		return &eval.TaggedValue{
-			CtorName: "ThinkingDelta",
-			Fields:   []eval.Value{&eval.StringValue{Value: c.Text}},
-		}
-	case ai.StreamUsage:
-		usageRec := &eval.RecordValue{
-			Fields: map[string]eval.Value{
-				"input_tokens":                &eval.IntValue{Value: c.InputTokens},
-				"output_tokens":               &eval.IntValue{Value: c.OutputTokens},
-				"cache_read_input_tokens":     &eval.IntValue{Value: c.CacheReadInputTokens},
-				"cache_creation_input_tokens": &eval.IntValue{Value: c.CacheCreationInputTokens},
-			},
-		}
-		return &eval.TaggedValue{
-			CtorName: "Usage",
-			Fields:   []eval.Value{usageRec},
-		}
-	default:
-		return nil
-	}
-}
-
 // ============================================================================
-// Decoders — AILANG records → Go structs
+// aiStepWithStreamRecorded — the recorded-stream API (#546), adopted from
+// @arniwesth's prototype (arniwesth/ailang#2) and productionized in v0.32.0
+// with the fail-loud + bounded-inert-drain policy (see ai_stream_core.go).
+//
+// Identical to aiStepWithStream except the return shape: it preserves
+// immediate per-chunk callbacks AND returns the exact ordered list of observed
+// chunks alongside the final outcome:
+//
+//   { chunks: [StreamChunk], outcome: Result[StepResult, AIError] }
+//
+// The chunks are returned on BOTH outcomes. That is the point of the shape: a
+// Result[{result, chunks}, AIError] discards every chunk observed before a
+// mid-stream failure, which is the case a deterministic replay most depends on.
 // ============================================================================
 
-// decodeMessages converts an AILANG list of Message records into []ai.Message.
-// Tolerates missing fields (treated as zero values) so callers can omit
-// optional fields like ToolCalls / ToolCallID. Rejects non-record entries.
-func decodeMessages(list *eval.ListValue) ([]ai.Message, error) {
-	out := make([]ai.Message, 0, len(list.Elements))
-	for i, elem := range list.Elements {
-		rec, ok := elem.(*eval.RecordValue)
-		if !ok {
-			return nil, fmt.Errorf("messages[%d]: expected record, got %T", i, elem)
-		}
-		msg := ai.Message{
-			Role:       getStringField(rec, "role"),
-			Content:    getStringField(rec, "content"),
-			ToolCallID: getStringField(rec, "tool_call_id"),
-		}
-		// tool_calls is an optional list[ToolCall]; absent or nil = no calls.
-		if tcList, ok := rec.Fields["tool_calls"].(*eval.ListValue); ok {
-			calls, err := decodeToolCalls(tcList)
-			if err != nil {
-				return nil, fmt.Errorf("messages[%d]: %w", i, err)
-			}
-			msg.ToolCalls = calls
-		}
-		// images is a list[ImagePart] (M-STD-AI-VISION-INPUT). Empty/absent =
-		// text-only message, wire-identical to pre-vision (Images stays nil).
-		if imgList, ok := rec.Fields["images"].(*eval.ListValue); ok && len(imgList.Elements) > 0 {
-			images, err := decodeImageParts(imgList)
-			if err != nil {
-				return nil, fmt.Errorf("messages[%d]: %w", i, err)
-			}
-			msg.Images = images
-		}
-		out = append(out, msg)
-	}
-	return out, nil
-}
-
-// decodeToolCalls converts an AILANG list of ToolCall records into []ai.ToolCall.
-func decodeToolCalls(list *eval.ListValue) ([]ai.ToolCall, error) {
-	out := make([]ai.ToolCall, 0, len(list.Elements))
-	for i, elem := range list.Elements {
-		rec, ok := elem.(*eval.RecordValue)
-		if !ok {
-			return nil, fmt.Errorf("tool_calls[%d]: expected record, got %T", i, elem)
-		}
-		out = append(out, ai.ToolCall{
-			ID:        getStringField(rec, "id"),
-			Name:      getStringField(rec, "name"),
-			Arguments: getStringField(rec, "arguments"),
-		})
-	}
-	return out, nil
-}
-
-// decodeImageParts decodes an AILANG list[ImagePart] into []ai.ImagePart.
-// Each element is a {source, mime} record (M-STD-AI-VISION-INPUT). An empty
-// source is rejected — the caller (decodeMessages) surfaces it as a typed
-// SchemaValidation AIError rather than silently forwarding a blank image.
-func decodeImageParts(list *eval.ListValue) ([]ai.ImagePart, error) {
-	out := make([]ai.ImagePart, 0, len(list.Elements))
-	for i, elem := range list.Elements {
-		rec, ok := elem.(*eval.RecordValue)
-		if !ok {
-			return nil, fmt.Errorf("images[%d]: expected record, got %T", i, elem)
-		}
-		source := getStringField(rec, "source")
-		if source == "" {
-			return nil, fmt.Errorf("images[%d]: empty source (expected base64 or data-URI)", i)
-		}
-		out = append(out, ai.ImagePart{
-			Source: source,
-			Mime:   getStringField(rec, "mime"),
-		})
-	}
-	return out, nil
-}
-
-// decodeCacheBreakpoints converts an AILANG list of CacheBreakpoint records
-// into []ai.CacheBreakpoint. Tolerates missing fields (treated as empty
-// strings) — provider-side validation rejects unknown positions with a
-// once-per-session warning rather than failing.
-func decodeCacheBreakpoints(list *eval.ListValue) ([]ai.CacheBreakpoint, error) {
-	if list == nil || len(list.Elements) == 0 {
-		return nil, nil
-	}
-	out := make([]ai.CacheBreakpoint, 0, len(list.Elements))
-	for i, elem := range list.Elements {
-		rec, ok := elem.(*eval.RecordValue)
-		if !ok {
-			return nil, fmt.Errorf("cache_breakpoints[%d]: expected record, got %T", i, elem)
-		}
-		out = append(out, ai.CacheBreakpoint{
-			Position: getStringField(rec, "position"),
-			TTL:      getStringField(rec, "ttl"),
-		})
-	}
-	return out, nil
-}
-
-// decodeToolSchemas converts an AILANG list of ToolSchema records into
-// []ai.ToolSchema.
-func decodeToolSchemas(list *eval.ListValue) ([]ai.ToolSchema, error) {
-	out := make([]ai.ToolSchema, 0, len(list.Elements))
-	for i, elem := range list.Elements {
-		rec, ok := elem.(*eval.RecordValue)
-		if !ok {
-			return nil, fmt.Errorf("tools[%d]: expected record, got %T", i, elem)
-		}
-		out = append(out, ai.ToolSchema{
-			Name:        getStringField(rec, "name"),
-			Description: getStringField(rec, "description"),
-			Parameters:  getStringField(rec, "parameters"),
-		})
-	}
-	return out, nil
-}
-
-// getStringField extracts a string field from a record, returning ""
-// if the field is missing or not a StringValue. Forgiving by design —
-// schema-level rigor is the type checker's job, not this layer.
-func getStringField(rec *eval.RecordValue, name string) string {
-	v, ok := rec.Fields[name]
-	if !ok {
-		return ""
+func aiStepWithStreamRecorded(ctx *EffContext, args []eval.Value) (eval.Value, error) {
+	recorded, resp, aiErr, err := aiStreamCore(ctx, args, "stepWithStreamRecorded", streamRecordPolicy{record: true, failLoud: true})
+	if err != nil {
+		return nil, err
 	}
-	s, ok := v.(*eval.StringValue)
-	if !ok {
-		return ""
+	if aiErr != nil {
+		return makeRecordedStream(recorded, makeAIErrorResultRecord(aiErr)), nil
 	}
-	return s.Value
+	return makeRecordedStream(recorded, makeOkStepResult(resp)), nil
 }
 
-// ============================================================================
-// Encoders — Go structs → AILANG Result records
-// ============================================================================
-
-// makeOkStringResult builds Ok(string) — for callResult / callJsonResult.
-func makeOkStringResult(s string) eval.Value {
-	return &eval.TaggedValue{
-		CtorName: "Ok",
-		Fields:   []eval.Value{&eval.StringValue{Value: s}},
+// makeRecordedStream builds { chunks: [StreamChunk], outcome: Result[...] }.
+func makeRecordedStream(chunks []eval.Value, outcome eval.Value) eval.Value {
+	if chunks == nil {
+		chunks = []eval.Value{}
 	}
-}
-
-// makeOkStepResult builds Ok(StepResult record) — for step.
-func makeOkStepResult(resp *ai.Response) eval.Value {
-	// Build the assistant Message record.
-	msgRec := &eval.RecordValue{
+	return &eval.RecordValue{
 		Fields: map[string]eval.Value{
-			"role":         &eval.StringValue{Value: "assistant"},
-			"content":      &eval.StringValue{Value: resp.Text},
-			"tool_calls":   encodeToolCalls(resp.ToolCalls),
-			"tool_call_id": &eval.StringValue{Value: ""},
-		},
-	}
-	stepResult := &eval.RecordValue{
-		Fields: map[string]eval.Value{
-			"message":                     msgRec,
-			"tool_calls":                  encodeToolCalls(resp.ToolCalls),
-			"finish_reason":               &eval.StringValue{Value: resp.FinishReason},
-			"input_tokens":                &eval.IntValue{Value: resp.InputTokens},
-			"output_tokens":               &eval.IntValue{Value: resp.OutputTokens},
-			"cache_read_input_tokens":     &eval.IntValue{Value: resp.CacheReadInputTokens},
-			"cache_creation_input_tokens": &eval.IntValue{Value: resp.CacheCreationInputTokens},
-		},
-	}
-	return &eval.TaggedValue{
-		CtorName: "Ok",
-		Fields:   []eval.Value{stepResult},
-	}
-}
-
-// encodeToolCalls builds an AILANG list[ToolCall] from a Go slice.
-func encodeToolCalls(calls []ai.ToolCall) eval.Value {
-	elems := make([]eval.Value, 0, len(calls))
-	for _, c := range calls {
-		elems = append(elems, &eval.RecordValue{
-			Fields: map[string]eval.Value{
-				"id":        &eval.StringValue{Value: c.ID},
-				"name":      &eval.StringValue{Value: c.Name},
-				"arguments": &eval.StringValue{Value: c.Arguments},
-			},
-		})
-	}
-	return &eval.ListValue{Elements: elems}
-}
-
-// makeAIErrorResultRecord builds Err(AIError record) for any AI op that
-// surfaces a typed failure. AIError shape: {code, message, retryable},
-// matching std/ai/streaming.AIError byte-for-byte.
-func makeAIErrorResultRecord(e *ai.AIError) eval.Value {
-	if e == nil {
-		// Defensive: should never happen, but if it does emit an
-		// internal-coded record so downstream consumers see something
-		// meaningful instead of a nil-deref.
-		e = ai.NewAIError(ai.CodeInternal, "nil AIError surfaced from effect op", false)
-	}
-	return &eval.TaggedValue{
-		CtorName: "Err",
-		Fields: []eval.Value{
-			&eval.RecordValue{
-				Fields: map[string]eval.Value{
-					"code":      &eval.StringValue{Value: e.Code},
-					"message":   &eval.StringValue{Value: e.Message},
-					"retryable": &eval.BoolValue{Value: e.Retryable},
-				},
-			},
+			"chunks":  &eval.ListValue{Elements: chunks},
+			"outcome": outcome,
 		},
 	}
 }
diff --git a/internal/effects/ai_step_with_stream_recorded_test.go b/internal/effects/ai_step_with_stream_recorded_test.go
new file mode 100644
index 000000000..78928b04c
--- /dev/null
+++ b/internal/effects/ai_step_with_stream_recorded_test.go
@@ -0,0 +1,244 @@
+package effects
+
+// Tests for aiStepWithStreamRecorded — the recorded-stream variant that keeps
+// immediate per-chunk callback delivery AND returns the exact ordered observed
+// chunks alongside the final outcome.
+//
+// The properties under test are the three the consumer depends on:
+//
+//   1. delivery AND capture, not either — the callback still fires per chunk
+//      while the same chunks come back in the result;
+//   2. chunks on BOTH outcomes — a stream that fails part-way still returns
+//      every chunk observed before the failure;
+//   3. identity, not reconstruction — the returned chunks are the values
+//      handed to the callback, in order, and concatenating the ContentDelta
+//      payloads still equals StepResult.message.content.
+//
+// The success-path fake is the shared fakeStepHandler from ai_step_test.go.
+// The error path needs its own: fakeStepHandler returns before emitting
+// anything when Step fails, so "chunks then error" was previously untestable.
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/sunholo-data/ailang/internal/ai"
+	"github.com/sunholo-data/ailang/internal/eval"
+)
+
+// partialThenFailHandler emits real chunks and *then* fails, which is the
+// case the recorded API exists for and the one a Result[{result, chunks}, ...]
+// shape would silently discard.
+type partialThenFailHandler struct {
+	chunks []string
+	err    error
+}
+
+func (h *partialThenFailHandler) Call(_ string) (string, error)        { return "", nil }
+func (h *partialThenFailHandler) CallJson(_, _ string) (string, error) { return "", nil }
+func (h *partialThenFailHandler) CallImage(_, out, _ string) (string, error) {
+	return out, nil
+}
+func (h *partialThenFailHandler) CallImageBase64(_, _ string) (string, error) { return "", nil }
+func (h *partialThenFailHandler) Step(_ string, _ []ai.Message, _ []ai.ToolSchema) (*ai.Response, error) {
+	return nil, h.err
+}
+func (h *partialThenFailHandler) StepWithCache(_ string, _ []ai.Message, _ []ai.ToolSchema, _ []ai.CacheBreakpoint) (*ai.Response, error) {
+	return nil, h.err
+}
+func (h *partialThenFailHandler) StepWithStream(_ string, _ []ai.Message, _ []ai.ToolSchema, _ []ai.CacheBreakpoint, onChunk func(ai.StreamChunk)) (*ai.Response, error) {
+	for _, c := range h.chunks {
+		onChunk(ai.StreamContentDelta{Text: c})
+	}
+	return nil, h.err
+}
+
+// recordedArgs builds the 5-argument call shape shared by these tests.
+func recordedArgs() []eval.Value {
+	return []eval.Value{
+		&eval.StringValue{Value: "gpt-4o"},
+		&eval.ListValue{Elements: []eval.Value{
+			&eval.RecordValue{Fields: map[string]eval.Value{
+				"role":       &eval.StringValue{Value: "user"},
+				"content":    &eval.StringValue{Value: "hi"},
+				"tool_calls": &eval.ListValue{Elements: []eval.Value{}},
+			}},
+		}},
+		&eval.ListValue{Elements: []eval.Value{}},
+		&eval.ListValue{Elements: []eval.Value{}},
+		&eval.UnitValue{},
+	}
+}
+
+// splitRecorded destructures the { chunks, outcome } record.
+func splitRecorded(t *testing.T, out eval.Value) ([]eval.Value, *eval.TaggedValue) {
+	t.Helper()
+	rec, ok := out.(*eval.RecordValue)
+	if !ok {
+		t.Fatalf("result type = %T, want *eval.RecordValue", out)
+	}
+	chunks, ok := rec.Fields["chunks"].(*eval.ListValue)
+	if !ok {
+		t.Fatalf("result.chunks type = %T, want *eval.ListValue", rec.Fields["chunks"])
+	}
+	outcome, ok := rec.Fields["outcome"].(*eval.TaggedValue)
+	if !ok {
+		t.Fatalf("result.outcome type = %T, want *eval.TaggedValue", rec.Fields["outcome"])
+	}
+	return chunks.Elements, outcome
+}
+
+func contentDeltaText(t *testing.T, v eval.Value) (string, bool) {
+	t.Helper()
+	tv, ok := v.(*eval.TaggedValue)
+	if !ok || tv.CtorName != "ContentDelta" {
+		return "", false
+	}
+	return tv.Fields[0].(*eval.StringValue).Value, true
+}
+
+// TestAIStepWithStreamRecorded_ReturnsDeliveredChunksOnSuccess is property 1
+// and property 3: the callback fired, and the identical sequence came back.
+func TestAIStepWithStreamRecorded_ReturnsDeliveredChunksOnSuccess(t *testing.T) {
+	h := &fakeStepHandler{
+		stepResp: &ai.Response{
+			Text: "hello world", InputTokens: 42, OutputTokens: 7, FinishReason: "stop",
+		},
+	}
+	var captured []eval.Value
+	ctx := &EffContext{
+		AI:       NewAIContext(h),
+		FnCaller: captureFnCaller(&captured),
+		Caps:     map[string]Capability{"AI": NewCapability("AI")},
+	}
+
+	out, err := aiStepWithStreamRecorded(ctx, recordedArgs())
+	if err != nil {
+		t.Fatalf("aiStepWithStreamRecorded returned Go error: %v", err)
+	}
+	returned, outcome := splitRecorded(t, out)
+
+	// Delivery still happened, and capture did not replace it.
+	if len(captured) != 2 {
+		t.Fatalf("callback invocations = %d, want 2", len(captured))
+	}
+	// No duplicate delivery: projected count == returned count.
+	if len(returned) != len(captured) {
+		t.Fatalf("returned chunk count = %d, want %d (same as delivered)", len(returned), len(captured))
+	}
+	// Identity, in order: the returned values are the delivered values.
+	for i := range captured {
+		if captured[i] != returned[i] {
+			t.Errorf("chunk %d: returned value is not the delivered value", i)
+		}
+	}
+	if outcome.CtorName != "Ok" {
+		t.Fatalf("outcome.CtorName = %q, want Ok", outcome.CtorName)
+	}
+}
+
+// TestAIStepWithStreamRecorded_ReturnsChunksOnErrorPath is property 2, and the
+// reason the return is a record rather than Result[{result, chunks}, AIError]:
+// that shape has nowhere to put chunks when the stream fails.
+func TestAIStepWithStreamRecorded_ReturnsChunksOnErrorPath(t *testing.T) {
+	h := &partialThenFailHandler{
+		chunks: []string{"partial-1", "partial-2"},
+		err:    errors.New("connection reset by peer"),
+	}
+	var captured []eval.Value
+	ctx := &EffContext{
+		AI:       NewAIContext(h),
+		FnCaller: captureFnCaller(&captured),
+		Caps:     map[string]Capability{"AI": NewCapability("AI")},
+	}
+
+	out, err := aiStepWithStreamRecorded(ctx, recordedArgs())
+	if err != nil {
+		t.Fatalf("aiStepWithStreamRecorded returned Go error: %v", err)
+	}
+	returned, outcome := splitRecorded(t, out)
+
+	if outcome.CtorName != "Err" {
+		t.Fatalf("outcome.CtorName = %q, want Err", outcome.CtorName)
+	}
+	// Both pre-failure chunks survive.
+	if len(returned) != 2 {
+		t.Fatalf("returned chunk count on error path = %d, want 2", len(returned))
+	}
+	want := []string{"partial-1", "partial-2"}
+	for i, w := range want {
+		got, ok := contentDeltaText(t, returned[i])
+		if !ok {
+			t.Fatalf("returned[%d] is not a ContentDelta", i)
+		}
+		if got != w {
+			t.Errorf("returned[%d] = %q, want %q", i, got, w)
+		}
+	}
+	if len(captured) != 2 {
+		t.Errorf("callback invocations on error path = %d, want 2", len(captured))
+	}
+}
+
+// TestAIStepWithStreamRecorded_ContentDeltaConcatEqualsMessageContent holds the
+// documented StreamChunk invariant across the new surface.
+func TestAIStepWithStreamRecorded_ContentDeltaConcatEqualsMessageContent(t *testing.T) {
+	h := &fakeStepHandler{
+		stepResp: &ai.Response{Text: "hello world", FinishReason: "stop"},
+	}
+	var captured []eval.Value
+	ctx := &EffContext{
+		AI:       NewAIContext(h),
+		FnCaller: captureFnCaller(&captured),
+		Caps:     map[string]Capability{"AI": NewCapability("AI")},
+	}
+
+	out, err := aiStepWithStreamRecorded(ctx, recordedArgs())
+	if err != nil {
+		t.Fatalf("aiStepWithStreamRecorded returned Go error: %v", err)
+	}
+	returned, outcome := splitRecorded(t, out)
+
+	concat := ""
+	for _, c := range returned {
+		if text, ok := contentDeltaText(t, c); ok {
+			concat += text
+		}
+	}
+	stepResult := outcome.Fields[0].(*eval.RecordValue)
+	msg := stepResult.Fields["message"].(*eval.RecordValue)
+	content := msg.Fields["content"].(*eval.StringValue).Value
+	if concat != content {
+		t.Errorf("concat(ContentDelta) = %q, want message.content %q", concat, content)
+	}
+}
+
+// TestAIStepWithStream_UnchangedByRecordedVariant pins the additive claim: the
+// existing entry point still returns Result[StepResult, AIError] directly, not
+// a record, so no current caller is affected.
+func TestAIStepWithStream_UnchangedByRecordedVariant(t *testing.T) {
+	h := &fakeStepHandler{
+		stepResp: &ai.Response{Text: "hello world", FinishReason: "stop"},
+	}
+	var captured []eval.Value
+	ctx := &EffContext{
+		AI:       NewAIContext(h),
+		FnCaller: captureFnCaller(&captured),
+		Caps:     map[string]Capability{"AI": NewCapability("AI")},
+	}
+
+	out, err := aiStepWithStream(ctx, recordedArgs())
+	if err != nil {
+		t.Fatalf("aiStepWithStream returned Go error: %v", err)
+	}
+	tagged, ok := out.(*eval.TaggedValue)
+	if !ok {
+		t.Fatalf("stepWithStream result type = %T, want *eval.TaggedValue (unchanged)", out)
+	}
+	if tagged.CtorName != "Ok" {
+		t.Fatalf("stepWithStream outcome = %q, want Ok", tagged.CtorName)
+	}
+	if len(captured) != 2 {
+		t.Errorf("stepWithStream callback invocations = %d, want 2", len(captured))
+	}
+}
diff --git a/internal/effects/ai_stream_core.go b/internal/effects/ai_stream_core.go
new file mode 100644
index 000000000..45ae37f74
--- /dev/null
+++ b/internal/effects/ai_stream_core.go
@@ -0,0 +1,165 @@
+package effects
+
+import (
+	"fmt"
+
+	"github.com/sunholo-data/ailang/internal/ai"
+	"github.com/sunholo-data/ailang/internal/eval"
+)
+
+type streamRecordPolicy struct {
+	record   bool
+	failLoud bool
+}
+
+const (
+	unencodableStreamChunkErrorPrefix = "unencodable stream chunk"
+	recordedDrainMaxChunks            = 256
+	recordedDrainMaxBytes             = 1 << 20
+)
+
+// aiStreamCore is the single validation, decode, dispatch, delivery, and trace
+// implementation shared by both streaming operations. Public wrappers retain
+// responsibility for constructing their intentionally different return shapes.
+func aiStreamCore(ctx *EffContext, args []eval.Value, opName string, policy streamRecordPolicy) ([]eval.Value, *ai.Response, *ai.AIError, error) {
+	model, messagesArg, toolsArg, breakpointsArg, onChunkFn, err := validateStreamArgs(args, opName)
+	if err != nil {
+		return nil, nil, nil, err
+	}
+	if ctx.AI == nil {
+		return nil, nil, ai.NewAIError(ai.CodeProviderNotFound, ErrNoAIHandler.Error(), false), nil
+	}
+	if ctx.FnCaller == nil {
+		return nil, nil, ai.NewAIError(ai.CodeInternal, opName+": FnCaller not wired (evaluator integration missing)", false), nil
+	}
+
+	messages, conversionErr := decodeMessages(messagesArg)
+	if conversionErr != nil {
+		return nil, nil, schemaAIError(conversionErr), nil
+	}
+	tools, conversionErr := decodeToolSchemas(toolsArg)
+	if conversionErr != nil {
+		return nil, nil, schemaAIError(conversionErr), nil
+	}
+	breakpoints, conversionErr := decodeCacheBreakpoints(breakpointsArg)
+	if conversionErr != nil {
+		return nil, nil, schemaAIError(conversionErr), nil
+	}
+
+	var recorded []eval.Value
+	if policy.record {
+		recorded = make([]eval.Value, 0, 16)
+	}
+	providerChunks := 0
+	postFailureChunks := 0
+	postFailureBytes := 0
+	fatalProviderIndex := 0
+	drainExhausted := false
+	var latchedErr *ai.AIError
+	onChunk := func(chunk ai.StreamChunk) {
+		if drainExhausted {
+			return
+		}
+		providerChunks++
+		if latchedErr != nil {
+			postFailureChunks++
+			payloadBytes := streamChunkPayloadBytes(chunk)
+			if payloadBytes >= recordedDrainMaxBytes-postFailureBytes {
+				postFailureBytes = recordedDrainMaxBytes
+			} else {
+				postFailureBytes += payloadBytes
+			}
+			if postFailureChunks >= recordedDrainMaxChunks || postFailureBytes >= recordedDrainMaxBytes {
+				drainExhausted = true
+			}
+			return
+		}
+		encoded := encodeStreamChunk(chunk)
+		if encoded == nil {
+			if policy.failLoud {
+				fatalProviderIndex = providerChunks
+				latchedErr = ai.NewAIError(ai.CodeInternal,
+					fmt.Sprintf("%s at provider index %d; recorded log is an incomplete prefix", unencodableStreamChunkErrorPrefix, fatalProviderIndex), false)
+			}
+			return
+		}
+		if policy.record {
+			recorded = append(recorded, encoded)
+		}
+		if _, callErr := ctx.FnCaller(onChunkFn, encoded); callErr != nil {
+			ctx.RecordAIEffect(opName+".callback",
+				[]string{fmt.Sprintf("chunk:%d", providerChunks)},
+				fmt.Sprintf(errResultPrefix, callErr.Error()), nil)
+		}
+	}
+
+	resp, stepErr := ctx.AI.StepWithStream(model.Value, messages, tools, breakpoints, onChunk)
+	var aiErr *ai.AIError
+	if latchedErr != nil {
+		aiErr = latchedErr
+	} else if stepErr != nil {
+		aiErr = classifyOpError(stepErr)
+	}
+	recordStreamTerminalTrace(ctx, opName, model.Value, len(messages), len(tools), len(breakpoints), providerChunks, len(recorded), fatalProviderIndex, drainExhausted, resp, aiErr)
+	return recorded, resp, aiErr, nil
+}
+
+// streamChunkPayloadBytes accounts for post-failure provider payload without
+// retaining or encoding it. Token usage chunks have no text/JSON payload.
+func streamChunkPayloadBytes(chunk ai.StreamChunk) int {
+	switch c := chunk.(type) {
+	case ai.StreamContentDelta:
+		return len(c.Text)
+	case ai.StreamThinkingDelta:
+		return len(c.Text)
+	default:
+		return 0
+	}
+}
+
+func validateStreamArgs(args []eval.Value, opName string) (*eval.StringValue, *eval.ListValue, *eval.ListValue, *eval.ListValue, eval.Value, error) {
+	if len(args) < 5 {
+		return nil, nil, nil, nil, nil, fmt.Errorf("E_AI_TYPE_ERROR: %s: expected 5 arguments, got %d", opName, len(args))
+	}
+	model, ok := args[0].(*eval.StringValue)
+	if !ok {
+		return nil, nil, nil, nil, nil, fmt.Errorf("E_AI_TYPE_ERROR: %s: expected string model, got %T", opName, args[0])
+	}
+	messages, ok := args[1].(*eval.ListValue)
+	if !ok {
+		return nil, nil, nil, nil, nil, fmt.Errorf("E_AI_TYPE_ERROR: %s: expected list[Message] messages, got %T", opName, args[1])
+	}
+	tools, ok := args[2].(*eval.ListValue)
+	if !ok {
+		return nil, nil, nil, nil, nil, fmt.Errorf("E_AI_TYPE_ERROR: %s: expected list[ToolSchema] tools, got %T", opName, args[2])
+	}
+	breakpoints, ok := args[3].(*eval.ListValue)
+	if !ok {
+		return nil, nil, nil, nil, nil, fmt.Errorf("E_AI_TYPE_ERROR: %s: expected list[CacheBreakpoint] cache_breakpoints, got %T", opName, args[3])
+	}
+	if args[4] == nil {
+		return nil, nil, nil, nil, nil, fmt.Errorf("E_AI_TYPE_ERROR: %s: on_chunk callback is nil", opName)
+	}
+	return model, messages, tools, breakpoints, args[4], nil
+}
+
+func schemaAIError(err error) *ai.AIError {
+	return ai.NewAIError(ai.CodeSchemaValidation, err.Error(), false)
+}
+
+func recordStreamTerminalTrace(ctx *EffContext, opName, model string, messageCount, toolCount, cacheCount, providerChunks, deliveredChunks, fatalProviderIndex int, drainExhausted bool, resp *ai.Response, aiErr *ai.AIError) {
+	counts := fmt.Sprintf("messages:%d tools:%d cache:%d provider_chunks:%d delivered_chunks:%d", messageCount, toolCount, cacheCount, providerChunks, deliveredChunks)
+	if fatalProviderIndex > 0 {
+		counts += fmt.Sprintf(" fatal_provider_index:%d", fatalProviderIndex)
+	}
+	if drainExhausted {
+		counts += " drain_exhausted:true"
+	}
+	result := ""
+	if aiErr != nil {
+		result = fmt.Sprintf(errResultPrefix, aiErr.Code)
+	} else {
+		result = fmt.Sprintf("text:%s tool_calls:%d finish:%s cache_read:%d cache_create:%d", truncateForTrace(resp.Text), len(resp.ToolCalls), resp.FinishReason, resp.CacheReadInputTokens, resp.CacheCreationInputTokens)
+	}
+	ctx.RecordAIEffect(opName, []string{truncateForTrace(model), counts}, result, ctx.AI.LastRoutingMetadata())
+}
diff --git a/internal/effects/ai_stream_core_test.go b/internal/effects/ai_stream_core_test.go
new file mode 100644
index 000000000..4480df1fe
--- /dev/null
+++ b/internal/effects/ai_stream_core_test.go
@@ -0,0 +1,342 @@
+package effects
+
+import (
+	"errors"
+	"os"
+	"reflect"
+	"strings"
+	"testing"
+
+	"github.com/sunholo-data/ailang/internal/ai"
+	"github.com/sunholo-data/ailang/internal/eval"
+	"github.com/sunholo-data/ailang/internal/trace"
+)
+
+type scriptedStreamHandler struct {
+	fakeStepHandler
+	chunks []ai.StreamChunk
+	resp   *ai.Response
+	err    error
+	route  *trace.ResolvedRoute
+}
+
+func (h *scriptedStreamHandler) StepWithStream(_ string, _ []ai.Message, _ []ai.ToolSchema, _ []ai.CacheBreakpoint, onChunk func(ai.StreamChunk)) (*ai.Response, error) {
+	for _, chunk := range h.chunks {
+		onChunk(chunk)
+	}
+	return h.resp, h.err
+}
+
+func (h *scriptedStreamHandler) LastRoutingMetadata() *trace.ResolvedRoute { return h.route }
+
+func streamTestContext(h AIHandler, captured *[]eval.Value) *EffContext {
+	return &EffContext{
+		AI: NewAIContext(h),
+		FnCaller: func(_ eval.Value, arg eval.Value) (eval.Value, error) {
+			*captured = append(*captured, arg)
+			return &eval.UnitValue{}, nil
+		},
+		Trace: trace.NewCollector(),
+	}
+}
+
+func streamErrRecord(t *testing.T, outcome *eval.TaggedValue) *eval.RecordValue {
+	t.Helper()
+	if outcome.CtorName != "Err" {
+		t.Fatalf("outcome = %s, want Err", outcome.CtorName)
+	}
+	return outcome.Fields[0].(*eval.RecordValue)
+}
+
+func streamSuccessResponse() *ai.Response {
+	return &ai.Response{Text: "done", FinishReason: "stop", InputTokens: 7, OutputTokens: 3}
+}
+
+func terminalAIEvent(t *testing.T, ctx *EffContext) *trace.EffectEvent {
+	t.Helper()
+	events := ctx.Trace.Events()
+	for i := len(events) - 1; i >= 0; i-- {
+		if events[i].Effect != nil && !strings.HasSuffix(events[i].Effect.OpName, ".callback") {
+			return events[i].Effect
+		}
+	}
+	t.Fatal("terminal AI trace event not found")
+	return nil
+}
+
+func TestAIStreamCoreMatrix(t *testing.T) {
+	rows := []struct {
+		name string
+		run  func(*testing.T)
+	}{
+		{"01_argument_type_decode_failure_parity", testStreamArgumentParity},
+		{"02_handler_and_fncaller_typed_errors", testStreamMissingIntegration},
+		{"03_callback_failure_is_fail_soft", testStreamCallbackFailure},
+		{"04_all_chunk_variants_full_usage_order", testStreamChunkVariants},
+		{"05_unencodable_first_and_middle", testStreamUnencodable},
+		{"06_independent_drain_budgets", testStreamDrainBudgets},
+		{"07_latched_error_not_overwritten", testStreamLatchedError},
+		{"08_empty_stream_success_and_error", testStreamEmpty},
+		{"09_stable_order_identity_no_duplicates", testStreamIdentity},
+		{"10_capability_and_budget_layer_parity", testStreamCapabilityContract},
+		{"11_trace_contract", testStreamTrace},
+		{"12_registry_public_type_and_metadata", testStreamSurfaceContract},
+		{"13_nested_recorded_stream_vm_shape", testStreamNestedValueShape},
+		{"14_adr009_ordering_gate", testStreamADR009Ordering},
+	}
+	if len(rows) != 14 {
+		t.Fatalf("matrix rows = %d, want 14", len(rows))
+	}
+	for _, row := range rows {
+		t.Run(row.name, row.run)
+	}
+}
+
+func testStreamArgumentParity(t *testing.T) {
+	cases := []struct {
+		name string
+		args []eval.Value
+	}{
+		{"arity", nil},
+		{"model", []eval.Value{&eval.IntValue{Value: 1}, &eval.ListValue{}, &eval.ListValue{}, &eval.ListValue{}, &eval.UnitValue{}}},
+		{"messages", []eval.Value{&eval.StringValue{}, &eval.IntValue{}, &eval.ListValue{}, &eval.ListValue{}, &eval.UnitValue{}}},
+		{"tools", []eval.Value{&eval.StringValue{}, &eval.ListValue{}, &eval.IntValue{}, &eval.ListValue{}, &eval.UnitValue{}}},
+		{"breakpoints", []eval.Value{&eval.StringValue{}, &eval.ListValue{}, &eval.ListValue{}, &eval.IntValue{}, &eval.UnitValue{}}},
+		{"callback", []eval.Value{&eval.StringValue{}, &eval.ListValue{}, &eval.ListValue{}, &eval.ListValue{}, nil}},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			for _, call := range []func(*EffContext, []eval.Value) (eval.Value, error){aiStepWithStream, aiStepWithStreamRecorded} {
+				if _, err := call(&EffContext{}, tc.args); err == nil || !strings.Contains(err.Error(), "E_AI_TYPE_ERROR") {
+					t.Fatalf("error = %v, want E_AI_TYPE_ERROR", err)
+				}
+			}
+		})
+	}
+	bad := recordedArgs()
+	bad[1] = &eval.ListValue{Elements: []eval.Value{&eval.IntValue{Value: 1}}}
+	for _, call := range []func(*EffContext, []eval.Value) (eval.Value, error){aiStepWithStream, aiStepWithStreamRecorded} {
+		out, err := call(&EffContext{AI: NewAIContext(&scriptedStreamHandler{resp: streamSuccessResponse()}), FnCaller: captureFnCaller(&[]eval.Value{})}, bad)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if _, ok := out.(*eval.RecordValue); ok {
+			_, outcome := splitRecorded(t, out)
+			streamErrRecord(t, outcome)
+		} else if out.(*eval.TaggedValue).CtorName != "Err" {
+			t.Fatal("legacy decode failure did not return Err")
+		}
+	}
+}
+
+func testStreamMissingIntegration(t *testing.T) {
+	legacy, _ := aiStepWithStream(&EffContext{}, recordedArgs())
+	if streamErrRecord(t, legacy.(*eval.TaggedValue)).Fields["code"].(*eval.StringValue).Value != ai.CodeProviderNotFound {
+		t.Fatal("legacy missing handler code mismatch")
+	}
+	recorded, _ := aiStepWithStreamRecorded(&EffContext{}, recordedArgs())
+	chunks, outcome := splitRecorded(t, recorded)
+	if len(chunks) != 0 || streamErrRecord(t, outcome).Fields["code"].(*eval.StringValue).Value != ai.CodeProviderNotFound {
+		t.Fatal("recorded missing handler contract mismatch")
+	}
+	ctx := &EffContext{AI: NewAIContext(&scriptedStreamHandler{})}
+	recorded, _ = aiStepWithStreamRecorded(ctx, recordedArgs())
+	chunks, outcome = splitRecorded(t, recorded)
+	if len(chunks) != 0 || streamErrRecord(t, outcome).Fields["code"].(*eval.StringValue).Value != ai.CodeInternal {
+		t.Fatal("recorded missing FnCaller contract mismatch")
+	}
+}
+
+func testStreamCallbackFailure(t *testing.T) {
+	h := &scriptedStreamHandler{chunks: []ai.StreamChunk{ai.StreamContentDelta{Text: "a"}, ai.StreamContentDelta{Text: "b"}}, resp: streamSuccessResponse()}
+	ctx := &EffContext{AI: NewAIContext(h), Trace: trace.NewCollector(), FnCaller: func(eval.Value, eval.Value) (eval.Value, error) { return nil, errors.New("callback broke") }}
+	out, err := aiStepWithStreamRecorded(ctx, recordedArgs())
+	if err != nil {
+		t.Fatal(err)
+	}
+	chunks, outcome := splitRecorded(t, out)
+	if len(chunks) != 2 || outcome.CtorName != "Ok" || len(ctx.Trace.Events()) != 3 {
+		t.Fatalf("chunks=%d outcome=%s events=%d", len(chunks), outcome.CtorName, len(ctx.Trace.Events()))
+	}
+}
+
+func testStreamChunkVariants(t *testing.T) {
+	usage := ai.StreamUsage{InputTokens: 1, OutputTokens: 2, CacheReadInputTokens: 3, CacheCreationInputTokens: 4}
+	h := &scriptedStreamHandler{chunks: []ai.StreamChunk{ai.StreamContentDelta{Text: "c"}, ai.StreamThinkingDelta{Text: "r"}, usage}, resp: streamSuccessResponse()}
+	var captured []eval.Value
+	out, _ := aiStepWithStreamRecorded(streamTestContext(h, &captured), recordedArgs())
+	chunks, _ := splitRecorded(t, out)
+	wantNames := []string{"ContentDelta", "ThinkingDelta", "Usage"}
+	for i, want := range wantNames {
+		if chunks[i].(*eval.TaggedValue).CtorName != want {
+			t.Fatalf("chunk %d ctor = %s", i, chunks[i].(*eval.TaggedValue).CtorName)
+		}
+	}
+	usageRec := chunks[2].(*eval.TaggedValue).Fields[0].(*eval.RecordValue)
+	for field, want := range map[string]int{"input_tokens": 1, "output_tokens": 2, "cache_read_input_tokens": 3, "cache_creation_input_tokens": 4} {
+		if usageRec.Fields[field].(*eval.IntValue).Value != want {
+			t.Errorf("%s mismatch", field)
+		}
+	}
+}
+
+func testStreamUnencodable(t *testing.T) {
+	for _, chunks := range [][]ai.StreamChunk{{nil, ai.StreamContentDelta{Text: "later"}}, {ai.StreamContentDelta{Text: "prefix"}, nil, ai.StreamContentDelta{Text: "later"}}} {
+		h := &scriptedStreamHandler{chunks: chunks, resp: streamSuccessResponse()}
+		var captured []eval.Value
+		out, _ := aiStepWithStreamRecorded(streamTestContext(h, &captured), recordedArgs())
+		returned, outcome := splitRecorded(t, out)
+		if len(returned) != len(captured) || len(returned) != len(chunks)-2 {
+			t.Fatalf("returned=%d captured=%d", len(returned), len(captured))
+		}
+		msg := streamErrRecord(t, outcome).Fields["message"].(*eval.StringValue).Value
+		if !strings.HasPrefix(msg, unencodableStreamChunkErrorPrefix) || !strings.Contains(msg, "incomplete prefix") {
+			t.Fatalf("message = %q", msg)
+		}
+	}
+}
+
+func testStreamDrainBudgets(t *testing.T) {
+	// The budget values are public contract (sprint plan M3): a silent change
+	// must fail here, not ship. Self-referential feeding alone cannot catch a
+	// budget regression (a 256->2 mutation survived it — iter-135).
+	if recordedDrainMaxChunks != 256 || recordedDrainMaxBytes != 1<<20 {
+		t.Fatalf("drain budget contract changed: chunks=%d bytes=%d", recordedDrainMaxChunks, recordedDrainMaxBytes)
+	}
+	chunkBudget := make([]ai.StreamChunk, recordedDrainMaxChunks+10)
+	chunkBudget[0] = nil
+	for i := 1; i < len(chunkBudget); i++ {
+		chunkBudget[i] = ai.StreamUsage{}
+	}
+	byteBudget := []ai.StreamChunk{nil, ai.StreamContentDelta{Text: strings.Repeat("x", recordedDrainMaxBytes/2)}, ai.StreamThinkingDelta{Text: strings.Repeat("y", recordedDrainMaxBytes/2)}}
+	for _, chunks := range [][]ai.StreamChunk{chunkBudget, byteBudget} {
+		ctx := streamTestContext(&scriptedStreamHandler{chunks: chunks, resp: streamSuccessResponse()}, &[]eval.Value{})
+		out, _ := aiStepWithStreamRecorded(ctx, recordedArgs())
+		_, outcome := splitRecorded(t, out)
+		streamErrRecord(t, outcome)
+		if event := terminalAIEvent(t, ctx); !strings.Contains(event.Args[1], "drain_exhausted:true") {
+			t.Fatalf("trace args = %q", event.Args)
+		}
+	}
+	// Under-budget control: a drain that stays inside both budgets must NOT
+	// report exhaustion (proves the exhaustion assertions above are informative).
+	under := []ai.StreamChunk{nil, ai.StreamUsage{}, ai.StreamUsage{}}
+	ctx := streamTestContext(&scriptedStreamHandler{chunks: under, resp: streamSuccessResponse()}, &[]eval.Value{})
+	out, _ := aiStepWithStreamRecorded(ctx, recordedArgs())
+	_, outcome := splitRecorded(t, out)
+	streamErrRecord(t, outcome)
+	if event := terminalAIEvent(t, ctx); strings.Contains(event.Args[1], "drain_exhausted:true") {
+		t.Fatalf("under-budget drain reported exhaustion: %q", event.Args)
+	}
+}
+
+func testStreamLatchedError(t *testing.T) {
+	for _, providerErr := range []error{nil, ai.NewAIError(ai.CodeAuthFailed, "later", false)} {
+		h := &scriptedStreamHandler{chunks: []ai.StreamChunk{nil}, resp: streamSuccessResponse(), err: providerErr}
+		out, _ := aiStepWithStreamRecorded(streamTestContext(h, &[]eval.Value{}), recordedArgs())
+		_, outcome := splitRecorded(t, out)
+		rec := streamErrRecord(t, outcome)
+		if rec.Fields["code"].(*eval.StringValue).Value != ai.CodeInternal || !strings.HasPrefix(rec.Fields["message"].(*eval.StringValue).Value, unencodableStreamChunkErrorPrefix) {
+			t.Fatal("latched representation error was overwritten")
+		}
+	}
+}
+
+func testStreamEmpty(t *testing.T) {
+	for _, providerErr := range []error{nil, errors.New("empty failed")} {
+		h := &scriptedStreamHandler{resp: streamSuccessResponse(), err: providerErr}
+		out, _ := aiStepWithStreamRecorded(streamTestContext(h, &[]eval.Value{}), recordedArgs())
+		chunks, outcome := splitRecorded(t, out)
+		if len(chunks) != 0 || (providerErr == nil) != (outcome.CtorName == "Ok") {
+			t.Fatalf("chunks=%d outcome=%s", len(chunks), outcome.CtorName)
+		}
+	}
+}
+
+func testStreamIdentity(t *testing.T) {
+	h := &scriptedStreamHandler{chunks: []ai.StreamChunk{ai.StreamContentDelta{Text: "1"}, ai.StreamThinkingDelta{Text: "2"}, ai.StreamUsage{}}, resp: streamSuccessResponse()}
+	var captured []eval.Value
+	out, _ := aiStepWithStreamRecorded(streamTestContext(h, &captured), recordedArgs())
+	returned, _ := splitRecorded(t, out)
+	for i := range returned {
+		if returned[i] != captured[i] {
+			t.Fatalf("chunk %d was encoded twice", i)
+		}
+	}
+}
+
+func testStreamCapabilityContract(t *testing.T) {
+	source, err := os.ReadFile("../builtins/ai_step.go")
+	if err != nil {
+		t.Fatal(err)
+	}
+	text := string(source)
+	for _, impl := range []string{"aiStepWithStreamImpl", "aiStepWithStreamRecordedImpl"} {
+		start := strings.Index(text, "func "+impl)
+		if start < 0 || !strings.Contains(text[start:start+220], `RequireCapWithBudget("AI", "")`) {
+			t.Fatalf("%s does not share AI capability/budget gate", impl)
+		}
+	}
+}
+
+func testStreamTrace(t *testing.T) {
+	ctx := streamTestContext(&scriptedStreamHandler{chunks: []ai.StreamChunk{ai.StreamContentDelta{Text: "ok"}, nil, ai.StreamUsage{}}, resp: streamSuccessResponse()}, &[]eval.Value{})
+	out, _ := aiStepWithStreamRecorded(ctx, recordedArgs())
+	returned, _ := splitRecorded(t, out)
+	event := terminalAIEvent(t, ctx)
+	if event.OpName != "stepWithStreamRecorded" || event.Result != "err:Internal" {
+		t.Fatalf("trace op/result = %s/%s", event.OpName, event.Result)
+	}
+	for _, part := range []string{"provider_chunks:3", "delivered_chunks:1", "fatal_provider_index:2"} {
+		if !strings.Contains(event.Args[1], part) {
+			t.Errorf("missing %s in %q", part, event.Args[1])
+		}
+	}
+	if len(returned) != 1 {
+		t.Fatal("delivered_chunks != len(recorded)")
+	}
+}
+
+func testStreamSurfaceContract(t *testing.T) {
+	for path, needles := range map[string][]string{
+		"../builtins/ai_step.go": {"_ai_step_with_stream_recorded", `Since:     "v0.32.0"`, "StabilityExperimental", "makeAIStepWithStreamRecordedType"},
+		"../../std/ai.ail":       {"export type RecordedStream", "export func stepWithStreamRecorded"},
+	} {
+		data, err := os.ReadFile(path)
+		if err != nil {
+			t.Fatal(err)
+		}
+		for _, needle := range needles {
+			if !strings.Contains(string(data), needle) {
+				t.Errorf("%s missing %q", path, needle)
+			}
+		}
+	}
+}
+
+func testStreamNestedValueShape(t *testing.T) {
+	h := &scriptedStreamHandler{chunks: []ai.StreamChunk{ai.StreamContentDelta{Text: "nested"}}, resp: streamSuccessResponse()}
+	out, _ := aiStepWithStreamRecorded(streamTestContext(h, &[]eval.Value{}), recordedArgs())
+	rec := out.(*eval.RecordValue)
+	if _, ok := rec.Fields["chunks"].(*eval.ListValue); !ok {
+		t.Fatal("chunks is not VM-convertible list shape")
+	}
+	if _, ok := rec.Fields["outcome"].(*eval.TaggedValue); !ok {
+		t.Fatal("outcome is not VM-convertible ADT shape")
+	}
+}
+
+func testStreamADR009Ordering(t *testing.T) {
+	for _, providerErr := range []error{nil, errors.New("partial terminal error")} {
+		h := &scriptedStreamHandler{chunks: []ai.StreamChunk{ai.StreamContentDelta{Text: "a"}, ai.StreamThinkingDelta{Text: "b"}, ai.StreamUsage{}}, resp: streamSuccessResponse(), err: providerErr}
+		var projected []eval.Value
+		out, _ := aiStepWithStreamRecorded(streamTestContext(h, &projected), recordedArgs())
+		returned, outcome := splitRecorded(t, out)
+		if !reflect.DeepEqual(returned, projected) || len(returned) != 3 {
+			t.Fatal("returned log differs from immediate ordered projection")
+		}
+		if (providerErr == nil) != (outcome.CtorName == "Ok") {
+			t.Fatalf("outcome = %s", outcome.CtorName)
+		}
+	}
+}
diff --git a/internal/pipeline/testdata/builtin_types.golden b/internal/pipeline/testdata/builtin_types.golden
index b7232aa4b..12492cd6d 100644
--- a/internal/pipeline/testdata/builtin_types.golden
+++ b/internal/pipeline/testdata/builtin_types.golden
@@ -18,6 +18,7 @@ _ai_call_stream : (string, string, string) -> Result[string, {code: string, mess
 _ai_step : (string, list[{content: string, images: list[{mime: string, source: string}], role: string, tool_call_id: string, tool_calls: list[{arguments: string, id: string, name: string}]}], list[{description: string, name: string, parameters: string}]) -> Result[{cache_creation_input_tokens: int, cache_read_input_tokens: int, finish_reason: string, input_tokens: int, message: {content: string, images: list[{mime: string, source: string}], role: string, tool_call_id: string, tool_calls: list[{arguments: string, id: string, name: string}]}, output_tokens: int, tool_calls: list[{arguments: string, id: string, name: string}]}, {code: string, message: string, retryable: bool}] ! {AI}
 _ai_step_with_cache : (string, list[{content: string, images: list[{mime: string, source: string}], role: string, tool_call_id: string, tool_calls: list[{arguments: string, id: string, name: string}]}], list[{description: string, name: string, parameters: string}], list[{position: string, ttl: string}]) -> Result[{cache_creation_input_tokens: int, cache_read_input_tokens: int, finish_reason: string, input_tokens: int, message: {content: string, images: list[{mime: string, source: string}], role: string, tool_call_id: string, tool_calls: list[{arguments: string, id: string, name: string}]}, output_tokens: int, tool_calls: list[{arguments: string, id: string, name: string}]}, {code: string, message: string, retryable: bool}] ! {AI}
 _ai_step_with_stream : (string, list[{content: string, images: list[{mime: string, source: string}], role: string, tool_call_id: string, tool_calls: list[{arguments: string, id: string, name: string}]}], list[{description: string, name: string, parameters: string}], list[{position: string, ttl: string}], StreamChunk -> ()) -> Result[{cache_creation_input_tokens: int, cache_read_input_tokens: int, finish_reason: string, input_tokens: int, message: {content: string, images: list[{mime: string, source: string}], role: string, tool_call_id: string, tool_calls: list[{arguments: string, id: string, name: string}]}, output_tokens: int, tool_calls: list[{arguments: string, id: string, name: string}]}, {code: string, message: string, retryable: bool}] ! {AI}
+_ai_step_with_stream_recorded : (string, list[{content: string, images: list[{mime: string, source: string}], role: string, tool_call_id: string, tool_calls: list[{arguments: string, id: string, name: string}]}], list[{description: string, name: string, parameters: string}], list[{position: string, ttl: string}], StreamChunk -> ()) -> {chunks: list[StreamChunk], outcome: Result[{cache_creation_input_tokens: int, cache_read_input_tokens: int, finish_reason: string, input_tokens: int, message: {content: string, images: list[{mime: string, source: string}], role: string, tool_call_id: string, tool_calls: list[{arguments: string, id: string, name: string}]}, output_tokens: int, tool_calls: list[{arguments: string, id: string, name: string}]}, {code: string, message: string, retryable: bool}]} ! {AI}
 _ai_stream_call : (string, string, string) -> Result[StreamConn, StreamErrorKind] ! {AI, Net, Stream}
 _array_append : (Array[a], a) -> Array[a]
 _array_empty : () -> Array[a]
diff --git a/std/ai.ail b/std/ai.ail
index 035091ba1..85bed3cc6 100644
--- a/std/ai.ail
+++ b/std/ai.ail
@@ -336,6 +336,31 @@ export func stepWithStream(
 ) -> Result[StepResult, AIError] ! {AI} =
   _ai_step_with_stream(model, messages, tools, cache_breakpoints, on_chunk)
 
+-- RecordedStream is what stepWithStreamRecorded returns: the exact ordered
+-- list of observed chunks plus the final typed outcome. `chunks` is populated
+-- on BOTH outcomes — a stream that fails part-way still returns every chunk
+-- observed before the failure, which is the case a deterministic recorder
+-- most depends on.
+export type RecordedStream = {
+  chunks: [StreamChunk],
+  outcome: Result[StepResult, AIError]
+}
+
+-- stepWithStreamRecorded: stepWithStream that also returns what it streamed.
+--
+-- The callback still fires immediately at each chunk's arrival, so live
+-- rendering is unchanged. The returned `chunks` is an exact, non-duplicating
+-- record of the same sequence, which lets a caller record a live exchange and
+-- replay it deterministically without widening the callback's effect row.
+export func stepWithStreamRecorded(
+  model: string,
+  messages: [Message],
+  tools: [ToolSchema],
+  cache_breakpoints: [CacheBreakpoint],
+  on_chunk: (StreamChunk) -> () ! {IO}
+) -> RecordedStream ! {AI} =
+  _ai_step_with_stream_recorded(model, messages, tools, cache_breakpoints, on_chunk)
+
 -- runTools: convenience loop driver for multi-turn tool dispatch.
 --
 -- Calls step in a loop until the model returns finish_reason != "tool_calls"