feat(effects): recorded-stream API S1 — adopt @arniwesth's prototype, shared core, fail-loud bounded drain (#546)

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:

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

internal/builtins/ai_step.go

@@ -376,7 +376,7 @@
 				{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]"},
 			},
 			Returns: "Result[StepResult, AIError]",
 			SeeAlso: []string{
@@ -451,6 +451,87 @@
 	)
 }
 
+// ============================================================================
+// _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",

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
+// ============================================================================

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},
+				},
+			},
+		},
+	}
+}