diff --git a/runtime/csharp/Prompty.Core.Tests/Model/Contracts/Events/FailureChunkConversionTests.cs b/runtime/csharp/Prompty.Core.Tests/Model/Contracts/Events/FailureChunkConversionTests.cs new file mode 100644 index 000000000..993d3abf5 --- /dev/null +++ b/runtime/csharp/Prompty.Core.Tests/Model/Contracts/Events/FailureChunkConversionTests.cs @@ -0,0 +1,126 @@ +// +#nullable enable + +using Xunit; + +#pragma warning disable IDE0130 +namespace Prompty.Core; +#pragma warning restore IDE0130 + + +public class FailureChunkConversionTests +{ + [Fact] + public void LoadYamlInput() + { + string yamlData = """ +failure: + outcome: indeterminate + message: "SSE stream error: connection reset" + +"""; + + var instance = FailureChunk.FromYaml(yamlData); + + Assert.NotNull(instance); + } + + [Fact] + public void LoadJsonInput() + { + string jsonData = """ +{ + "failure": { + "outcome": "indeterminate", + "message": "SSE stream error: connection reset" + } +} +"""; + + var instance = FailureChunk.FromJson(jsonData); + Assert.NotNull(instance); + } + + [Fact] + public void RoundtripJson() + { + // Test that FromJson -> ToJson -> FromJson produces equivalent data + string jsonData = """ +{ + "failure": { + "outcome": "indeterminate", + "message": "SSE stream error: connection reset" + } +} +"""; + + var original = FailureChunk.FromJson(jsonData); + Assert.NotNull(original); + + var json = original.ToJson(); + Assert.False(string.IsNullOrEmpty(json)); + + var reloaded = FailureChunk.FromJson(json); + Assert.NotNull(reloaded); + } + + [Fact] + public void RoundtripYaml() + { + // Test that FromYaml -> ToYaml -> FromYaml produces equivalent data + string yamlData = """ +failure: + outcome: indeterminate + message: "SSE stream error: connection reset" + +"""; + + var original = FailureChunk.FromYaml(yamlData); + Assert.NotNull(original); + + var yaml = original.ToYaml(); + Assert.False(string.IsNullOrEmpty(yaml)); + + var reloaded = FailureChunk.FromYaml(yaml); + Assert.NotNull(reloaded); + } + + [Fact] + public void ToJsonProducesValidJson() + { + string jsonData = """ +{ + "failure": { + "outcome": "indeterminate", + "message": "SSE stream error: connection reset" + } +} +"""; + + var instance = FailureChunk.FromJson(jsonData); + var json = instance.ToJson(); + + // Verify it's valid JSON by parsing it + var parsed = System.Text.Json.JsonDocument.Parse(json); + Assert.NotNull(parsed); + } + + [Fact] + public void ToYamlProducesValidYaml() + { + string yamlData = """ +failure: + outcome: indeterminate + message: "SSE stream error: connection reset" + +"""; + + var instance = FailureChunk.FromYaml(yamlData); + var yaml = instance.ToYaml(); + + // Verify it's valid YAML by parsing it + var deserializer = new YamlDotNet.Serialization.DeserializerBuilder().Build(); + var parsed = deserializer.Deserialize(yaml); + Assert.NotNull(parsed); + } +} diff --git a/runtime/csharp/Prompty.Core.Tests/Model/Contracts/Events/StreamFailureConversionTests.cs b/runtime/csharp/Prompty.Core.Tests/Model/Contracts/Events/StreamFailureConversionTests.cs new file mode 100644 index 000000000..d69c1ae83 --- /dev/null +++ b/runtime/csharp/Prompty.Core.Tests/Model/Contracts/Events/StreamFailureConversionTests.cs @@ -0,0 +1,125 @@ +// +#nullable enable + +using Xunit; + +#pragma warning disable IDE0130 +namespace Prompty.Core; +#pragma warning restore IDE0130 + + +public class StreamFailureConversionTests +{ + [Fact] + public void LoadYamlInput() + { + string yamlData = """ +outcome: indeterminate +message: "SSE stream error: connection reset" + +"""; + + var instance = StreamFailure.FromYaml(yamlData); + + Assert.NotNull(instance); + Assert.Equal(StreamFailureOutcome.Indeterminate, instance.Outcome); + Assert.Equal("SSE stream error: connection reset", instance.Message); + } + + [Fact] + public void LoadJsonInput() + { + string jsonData = """ +{ + "outcome": "indeterminate", + "message": "SSE stream error: connection reset" +} +"""; + + var instance = StreamFailure.FromJson(jsonData); + Assert.NotNull(instance); + Assert.Equal(StreamFailureOutcome.Indeterminate, instance.Outcome); + Assert.Equal("SSE stream error: connection reset", instance.Message); + } + + [Fact] + public void RoundtripJson() + { + // Test that FromJson -> ToJson -> FromJson produces equivalent data + string jsonData = """ +{ + "outcome": "indeterminate", + "message": "SSE stream error: connection reset" +} +"""; + + var original = StreamFailure.FromJson(jsonData); + Assert.NotNull(original); + + var json = original.ToJson(); + Assert.False(string.IsNullOrEmpty(json)); + + var reloaded = StreamFailure.FromJson(json); + Assert.NotNull(reloaded); + Assert.Equal(StreamFailureOutcome.Indeterminate, reloaded.Outcome); + Assert.Equal("SSE stream error: connection reset", reloaded.Message); + } + + [Fact] + public void RoundtripYaml() + { + // Test that FromYaml -> ToYaml -> FromYaml produces equivalent data + string yamlData = """ +outcome: indeterminate +message: "SSE stream error: connection reset" + +"""; + + var original = StreamFailure.FromYaml(yamlData); + Assert.NotNull(original); + + var yaml = original.ToYaml(); + Assert.False(string.IsNullOrEmpty(yaml)); + + var reloaded = StreamFailure.FromYaml(yaml); + Assert.NotNull(reloaded); + Assert.Equal(StreamFailureOutcome.Indeterminate, reloaded.Outcome); + Assert.Equal("SSE stream error: connection reset", reloaded.Message); + } + + [Fact] + public void ToJsonProducesValidJson() + { + string jsonData = """ +{ + "outcome": "indeterminate", + "message": "SSE stream error: connection reset" +} +"""; + + var instance = StreamFailure.FromJson(jsonData); + var json = instance.ToJson(); + + // Verify it's valid JSON by parsing it + var parsed = System.Text.Json.JsonDocument.Parse(json); + Assert.NotNull(parsed); + } + + [Fact] + public void ToYamlProducesValidYaml() + { + string yamlData = """ +outcome: indeterminate +message: "SSE stream error: connection reset" + +"""; + + var instance = StreamFailure.FromYaml(yamlData); + var yaml = instance.ToYaml(); + + // Verify it's valid YAML by parsing it + var deserializer = new YamlDotNet.Serialization.DeserializerBuilder().Build(); + var parsed = deserializer.Deserialize(yaml); + Assert.NotNull(parsed); + } +} diff --git a/runtime/csharp/Prompty.Core/Model/Contracts/Events/FailureChunk.cs b/runtime/csharp/Prompty.Core/Model/Contracts/Events/FailureChunk.cs new file mode 100644 index 000000000..1d5bb33f8 --- /dev/null +++ b/runtime/csharp/Prompty.Core/Model/Contracts/Events/FailureChunk.cs @@ -0,0 +1,175 @@ +// +// Copyright (c) Microsoft. All rights reserved. +#nullable enable + +using System.Linq; +using System.Text.Json; +using System.Threading; +using YamlDotNet.Serialization; + +#pragma warning disable IDE0130 +namespace Prompty.Core; +#pragma warning restore IDE0130 + + /// + /// A classified failure chunk from the LLM response stream. + /// +public partial class FailureChunk : StreamChunk +{ + /// + /// The shorthand property name for this type, if any. + /// + public new static string? ShorthandProperty => null; + + /// + /// Initializes a new instance of . + /// +#pragma warning disable CS8618 + public FailureChunk() + { + } +#pragma warning restore CS8618 + + /// + /// The kind identifier for classified failure chunks + /// + public override string Kind { get; set; } = "failure"; + + /// + /// The classified stream failure + /// + public StreamFailure Failure { get; set; } + + + + #region Load Methods + + /// + /// Load a FailureChunk instance from a dictionary. + /// + /// The dictionary containing the data. + /// Optional context with pre/post processing callbacks. + /// The loaded FailureChunk instance. + public new static FailureChunk Load(Dictionary data, LoadContext? context = null) + { + context ??= new LoadContext(); + if (context is not null) + { + data = context.ProcessInput(data); + } + + + if (!data.TryGetValue("failure", out var requiredFailureValue) || requiredFailureValue is null) + { + throw new ArgumentException($"{context!.At("failure").Path}: missing required field"); + } + + // Create new instance + var instance = new FailureChunk(); + + + if (data.TryGetValue("kind", out var kindValue) && kindValue is not null) + { + instance.Kind = kindValue.ToString()!; + } + + if (data.TryGetValue("failure", out var failureValue) && failureValue is not null) + { + instance.Failure = StreamFailure.Load(failureValue.GetDictionary(StreamFailure.ShorthandProperty), context!.At("failure")); + } + + if (context is not null) + { + instance = context.ProcessOutput(instance); + } + return instance; + } + + + #endregion + + #region Save Methods + + /// + /// Save the FailureChunk instance to a dictionary. + /// + /// Optional context with pre/post processing callbacks. + /// The dictionary representation of this instance. + public override Dictionary Save(SaveContext? context = null) + { + var obj = this; + if (context is not null) + { + obj = context.ProcessObject(obj); + } + + + // Start with parent class properties + var result = base.Save(context); + + + result["kind"] = obj.Kind; + + + result["failure"] = obj.Failure?.Save(context); + + + return result; + } + + + /// + /// Convert the FailureChunk instance to a YAML string. + /// + /// Optional context with pre/post processing callbacks. + /// The YAML string representation of this instance. + public new string ToYaml(SaveContext? context = null) + { + context ??= new SaveContext(); + return context.ToYaml(Save(context)); + } + + /// + /// Convert the FailureChunk instance to a JSON string. + /// + /// Optional context with pre/post processing callbacks. + /// Whether to indent the output. Defaults to true. + /// The JSON string representation of this instance. + public new string ToJson(SaveContext? context = null, bool indent = true) + { + context ??= new SaveContext(); + return context.ToJson(Save(context), indent); + } + + /// + /// Load a FailureChunk instance from a JSON string. + /// + /// The JSON string to parse. + /// Optional context with pre/post processing callbacks. + /// The loaded FailureChunk instance. + public new static FailureChunk FromJson(string json, LoadContext? context = null) + { + using var doc = JsonDocument.Parse(json); + Dictionary dict; + dict = JsonSerializer.Deserialize>(json, JsonUtils.Options) + ?? throw new ArgumentException("Failed to parse JSON as dictionary"); + + return Load(dict, context); + } + + /// + /// Load a FailureChunk instance from a YAML string. + /// + /// The YAML string to parse. + /// Optional context with pre/post processing callbacks. + /// The loaded FailureChunk instance. + public new static FailureChunk FromYaml(string yaml, LoadContext? context = null) + { + var dict = YamlUtils.Deserializer.Deserialize>(yaml) + ?? throw new ArgumentException("Failed to parse YAML as dictionary"); + + return Load(dict, context); + } + + #endregion +} diff --git a/runtime/csharp/Prompty.Core/Model/Contracts/Events/StreamChunk.cs b/runtime/csharp/Prompty.Core/Model/Contracts/Events/StreamChunk.cs index 70cd30716..c640124b3 100644 --- a/runtime/csharp/Prompty.Core/Model/Contracts/Events/StreamChunk.cs +++ b/runtime/csharp/Prompty.Core/Model/Contracts/Events/StreamChunk.cs @@ -90,6 +90,7 @@ private static StreamChunk LoadKind(Dictionary data, LoadContex "tool" => ToolChunk.Load(data, context), "usage" => UsageChunk.Load(data, context), "error" => ErrorChunk.Load(data, context), + "failure" => FailureChunk.Load(data, context), _ => throw new ArgumentException($"Unknown StreamChunk discriminator field 'kind' value: {discriminator}"), }; diff --git a/runtime/csharp/Prompty.Core/Model/Contracts/Events/StreamFailure.cs b/runtime/csharp/Prompty.Core/Model/Contracts/Events/StreamFailure.cs new file mode 100644 index 000000000..2fc3eeaca --- /dev/null +++ b/runtime/csharp/Prompty.Core/Model/Contracts/Events/StreamFailure.cs @@ -0,0 +1,174 @@ +// +// Copyright (c) Microsoft. All rights reserved. +#nullable enable + +using System.Linq; +using System.Text.Json; +using System.Threading; +using YamlDotNet.Serialization; + +#pragma warning disable IDE0130 +namespace Prompty.Core; +#pragma warning restore IDE0130 + + /// + /// A classified terminal failure from an LLM response stream. + /// +public partial class StreamFailure +{ + /// + /// The shorthand property name for this type, if any. + /// + public static string? ShorthandProperty => null; + + /// + /// Initializes a new instance of . + /// +#pragma warning disable CS8618 + public StreamFailure() + { + } +#pragma warning restore CS8618 + + /// + /// Whether the provider outcome is known or requires reconciliation + /// + public StreamFailureOutcome Outcome { get; set; } = StreamFailureOutcome.Determinate; + + /// + /// The human-readable failure message + /// + public string Message { get; set; } = string.Empty; + + + + #region Load Methods + + /// + /// Load a StreamFailure instance from a dictionary. + /// + /// The dictionary containing the data. + /// Optional context with pre/post processing callbacks. + /// The loaded StreamFailure instance. + public static StreamFailure Load(Dictionary data, LoadContext? context = null) + { + context ??= new LoadContext(); + if (context is not null) + { + data = context.ProcessInput(data); + } + + + // Create new instance + var instance = new StreamFailure(); + + + if (data.TryGetValue("outcome", out var outcomeValue) && outcomeValue is not null) + { + instance.Outcome = StreamFailureOutcomeParser.Parse(outcomeValue.ToString()!); + } + + if (data.TryGetValue("message", out var messageValue) && messageValue is not null) + { + instance.Message = messageValue.ToString()!; + } + + if (context is not null) + { + instance = context.ProcessOutput(instance); + } + return instance; + } + + + #endregion + + #region Save Methods + + /// + /// Save the StreamFailure instance to a dictionary. + /// + /// Optional context with pre/post processing callbacks. + /// The dictionary representation of this instance. + public Dictionary Save(SaveContext? context = null) + { + var obj = this; + if (context is not null) + { + obj = context.ProcessObject(obj); + } + + + var result = new Dictionary(); + + + result["outcome"] = StreamFailureOutcomeParser.ToValue(obj.Outcome); + + + result["message"] = obj.Message; + + + if (context is not null) + { + result = context.ProcessDict(result); + } + + return result; + } + + + /// + /// Convert the StreamFailure instance to a YAML string. + /// + /// Optional context with pre/post processing callbacks. + /// The YAML string representation of this instance. + public string ToYaml(SaveContext? context = null) + { + context ??= new SaveContext(); + return context.ToYaml(Save(context)); + } + + /// + /// Convert the StreamFailure instance to a JSON string. + /// + /// Optional context with pre/post processing callbacks. + /// Whether to indent the output. Defaults to true. + /// The JSON string representation of this instance. + public string ToJson(SaveContext? context = null, bool indent = true) + { + context ??= new SaveContext(); + return context.ToJson(Save(context), indent); + } + + /// + /// Load a StreamFailure instance from a JSON string. + /// + /// The JSON string to parse. + /// Optional context with pre/post processing callbacks. + /// The loaded StreamFailure instance. + public static StreamFailure FromJson(string json, LoadContext? context = null) + { + using var doc = JsonDocument.Parse(json); + Dictionary dict; + dict = JsonSerializer.Deserialize>(json, JsonUtils.Options) + ?? throw new ArgumentException("Failed to parse JSON as dictionary"); + + return Load(dict, context); + } + + /// + /// Load a StreamFailure instance from a YAML string. + /// + /// The YAML string to parse. + /// Optional context with pre/post processing callbacks. + /// The loaded StreamFailure instance. + public static StreamFailure FromYaml(string yaml, LoadContext? context = null) + { + var dict = YamlUtils.Deserializer.Deserialize>(yaml) + ?? throw new ArgumentException("Failed to parse YAML as dictionary"); + + return Load(dict, context); + } + + #endregion +} diff --git a/runtime/csharp/Prompty.Core/Model/Contracts/Events/StreamFailureOutcome.cs b/runtime/csharp/Prompty.Core/Model/Contracts/Events/StreamFailureOutcome.cs new file mode 100644 index 000000000..c2914ba7c --- /dev/null +++ b/runtime/csharp/Prompty.Core/Model/Contracts/Events/StreamFailureOutcome.cs @@ -0,0 +1,43 @@ +// +// +// Code generated by Typra emitter; DO NOT EDIT. +#nullable enable + +using System; +using System.Text.Json.Serialization; + +namespace Prompty.Core; + +[JsonConverter(typeof(JsonStringEnumConverter))] +public enum StreamFailureOutcome +{ + [JsonPropertyName("determinate")] + Determinate, + + [JsonPropertyName("indeterminate")] + Indeterminate, + +} + +public static class StreamFailureOutcomeParser +{ + public static StreamFailureOutcome Parse(string value) + { + return value switch + { + "determinate" => StreamFailureOutcome.Determinate, + "indeterminate" => StreamFailureOutcome.Indeterminate, + _ => Enum.Parse(value, true), + }; + } + + public static string ToValue(StreamFailureOutcome value) + { + return value switch + { + StreamFailureOutcome.Determinate => "determinate", + StreamFailureOutcome.Indeterminate => "indeterminate", + _ => value.ToString().ToLowerInvariant(), + }; + } +} diff --git a/runtime/go/prompty/model/failure_chunk_test.go b/runtime/go/prompty/model/failure_chunk_test.go new file mode 100644 index 000000000..0a7338f1b --- /dev/null +++ b/runtime/go/prompty/model/failure_chunk_test.go @@ -0,0 +1,244 @@ +// +// Code generated by Typra emitter; DO NOT EDIT. + +package prompty_test + +import ( + "encoding/json" + "testing" + + "gopkg.in/yaml.v3" + + "prompty/model" +) + +// TestFailureChunkLoadJSON tests loading FailureChunk from JSON +func TestFailureChunkLoadJSON(t *testing.T) { + jsonData := ` +{ + "failure": { + "outcome": "indeterminate", + "message": "SSE stream error: connection reset" + } +} +` + var data map[string]interface{} + if err := json.Unmarshal([]byte(jsonData), &data); err != nil { + t.Fatalf("Failed to parse JSON: %v", err) + } + + ctx := prompty.NewLoadContext() + instance, err := prompty.LoadFailureChunk(data, ctx) + if err != nil { + t.Fatalf("Failed to load FailureChunk: %v", err) + } + _ = instance // No scalar properties to validate + if instance.Failure.Outcome != "indeterminate" { + t.Errorf(`Expected Failure.Outcome to be "indeterminate", got %v`, instance.Failure.Outcome) + } + if instance.Failure.Message != "SSE stream error: connection reset" { + t.Errorf(`Expected Failure.Message to be "SSE stream error: connection reset", got %v`, instance.Failure.Message) + } +} + +// TestFailureChunkLoadYAML tests loading FailureChunk from YAML +func TestFailureChunkLoadYAML(t *testing.T) { + yamlData := ` +failure: + outcome: indeterminate + message: "SSE stream error: connection reset" + +` + var data map[string]interface{} + if err := yaml.Unmarshal([]byte(yamlData), &data); err != nil { + t.Fatalf("Failed to parse YAML: %v", err) + } + + ctx := prompty.NewLoadContext() + instance, err := prompty.LoadFailureChunk(data, ctx) + if err != nil { + t.Fatalf("Failed to load FailureChunk: %v", err) + } + _ = instance // No scalar properties to validate + if instance.Failure.Outcome != "indeterminate" { + t.Errorf(`Expected Failure.Outcome to be "indeterminate", got %v`, instance.Failure.Outcome) + } + if instance.Failure.Message != "SSE stream error: connection reset" { + t.Errorf(`Expected Failure.Message to be "SSE stream error: connection reset", got %v`, instance.Failure.Message) + } +} + +// TestFailureChunkFromJSON tests loading FailureChunk through the generated JSON helper +func TestFailureChunkFromJSON(t *testing.T) { + jsonData := ` +{ + "failure": { + "outcome": "indeterminate", + "message": "SSE stream error: connection reset" + } +} +` + + instance, err := prompty.FailureChunkFromJSON(jsonData) + if err != nil { + t.Fatalf("Failed to load FailureChunk from JSON helper: %v", err) + } + _ = instance // No scalar properties to validate + if instance.Failure.Outcome != "indeterminate" { + t.Errorf(`Expected Failure.Outcome to be "indeterminate", got %v`, instance.Failure.Outcome) + } + if instance.Failure.Message != "SSE stream error: connection reset" { + t.Errorf(`Expected Failure.Message to be "SSE stream error: connection reset", got %v`, instance.Failure.Message) + } +} + +// TestFailureChunkFromYAML tests loading FailureChunk through the generated YAML helper +func TestFailureChunkFromYAML(t *testing.T) { + yamlData := ` +failure: + outcome: indeterminate + message: "SSE stream error: connection reset" + +` + + instance, err := prompty.FailureChunkFromYAML(yamlData) + if err != nil { + t.Fatalf("Failed to load FailureChunk from YAML helper: %v", err) + } + _ = instance // No scalar properties to validate + if instance.Failure.Outcome != "indeterminate" { + t.Errorf(`Expected Failure.Outcome to be "indeterminate", got %v`, instance.Failure.Outcome) + } + if instance.Failure.Message != "SSE stream error: connection reset" { + t.Errorf(`Expected Failure.Message to be "SSE stream error: connection reset", got %v`, instance.Failure.Message) + } +} + +// TestFailureChunkRoundtrip tests load -> save -> load produces equivalent data +func TestFailureChunkRoundtrip(t *testing.T) { + jsonData := ` +{ + "failure": { + "outcome": "indeterminate", + "message": "SSE stream error: connection reset" + } +} +` + var data map[string]interface{} + if err := json.Unmarshal([]byte(jsonData), &data); err != nil { + t.Fatalf("Failed to parse JSON: %v", err) + } + + loadCtx := prompty.NewLoadContext() + instance, err := prompty.LoadFailureChunk(data, loadCtx) + if err != nil { + t.Fatalf("Failed to load FailureChunk: %v", err) + } + saveCtx := prompty.NewSaveContext() + savedData := instance.Save(saveCtx) + + reloaded, err := prompty.LoadFailureChunk(savedData, loadCtx) + if err != nil { + t.Fatalf("Failed to reload FailureChunk: %v", err) + } + _ = reloaded // No scalar properties to validate + if reloaded.Failure.Outcome != "indeterminate" { + t.Errorf(`Expected Failure.Outcome to be "indeterminate", got %v`, reloaded.Failure.Outcome) + } + if reloaded.Failure.Message != "SSE stream error: connection reset" { + t.Errorf(`Expected Failure.Message to be "SSE stream error: connection reset", got %v`, reloaded.Failure.Message) + } +} + +// TestFailureChunkToJSON tests that ToJSON produces valid JSON +func TestFailureChunkToJSON(t *testing.T) { + jsonData := ` +{ + "failure": { + "outcome": "indeterminate", + "message": "SSE stream error: connection reset" + } +} +` + var data map[string]interface{} + if err := json.Unmarshal([]byte(jsonData), &data); err != nil { + t.Fatalf("Failed to parse JSON: %v", err) + } + + ctx := prompty.NewLoadContext() + instance, err := prompty.LoadFailureChunk(data, ctx) + if err != nil { + t.Fatalf("Failed to load FailureChunk: %v", err) + } + jsonOutput, err := instance.ToJSON() + if err != nil { + t.Fatalf("Failed to convert to JSON: %v", err) + } + + var parsed map[string]interface{} + if err := json.Unmarshal([]byte(jsonOutput), &parsed); err != nil { + t.Fatalf("Failed to parse generated JSON: %v", err) + } + + reloaded, err := prompty.LoadFailureChunk(parsed, ctx) + if err != nil { + t.Fatalf("Failed to reload generated JSON: %v", err) + } + _ = reloaded // No scalar properties to validate + if reloaded.Failure.Outcome != "indeterminate" { + t.Errorf(`Expected Failure.Outcome to be "indeterminate", got %v`, reloaded.Failure.Outcome) + } + if reloaded.Failure.Message != "SSE stream error: connection reset" { + t.Errorf(`Expected Failure.Message to be "SSE stream error: connection reset", got %v`, reloaded.Failure.Message) + } +} + +// TestFailureChunkToYAML tests that ToYAML produces valid YAML +func TestFailureChunkToYAML(t *testing.T) { + jsonData := ` +{ + "failure": { + "outcome": "indeterminate", + "message": "SSE stream error: connection reset" + } +} +` + var data map[string]interface{} + if err := json.Unmarshal([]byte(jsonData), &data); err != nil { + t.Fatalf("Failed to parse JSON: %v", err) + } + + ctx := prompty.NewLoadContext() + instance, err := prompty.LoadFailureChunk(data, ctx) + if err != nil { + t.Fatalf("Failed to load FailureChunk: %v", err) + } + yamlOutput, err := instance.ToYAML() + if err != nil { + t.Fatalf("Failed to convert to YAML: %v", err) + } + + var parsed map[string]interface{} + if err := yaml.Unmarshal([]byte(yamlOutput), &parsed); err != nil { + t.Fatalf("Failed to parse generated YAML: %v", err) + } + + reloaded, err := prompty.LoadFailureChunk(parsed, ctx) + if err != nil { + t.Fatalf("Failed to reload generated YAML: %v", err) + } + _ = reloaded // No scalar properties to validate + if reloaded.Failure.Outcome != "indeterminate" { + t.Errorf(`Expected Failure.Outcome to be "indeterminate", got %v`, reloaded.Failure.Outcome) + } + if reloaded.Failure.Message != "SSE stream error: connection reset" { + t.Errorf(`Expected Failure.Message to be "SSE stream error: connection reset", got %v`, reloaded.Failure.Message) + } +} + +// TestFailureChunkFromJSONInvalid rejects malformed JSON instead of silently defaulting +func TestFailureChunkFromJSONInvalid(t *testing.T) { + if _, err := prompty.FailureChunkFromJSON("{"); err == nil { + t.Fatalf("Expected malformed JSON to fail") + } +} diff --git a/runtime/go/prompty/model/stream_chunk.go b/runtime/go/prompty/model/stream_chunk.go index c2956c692..704e5cd54 100644 --- a/runtime/go/prompty/model/stream_chunk.go +++ b/runtime/go/prompty/model/stream_chunk.go @@ -43,6 +43,8 @@ func LoadStreamChunk(data interface{}, ctx *LoadContext) (interface{}, error) { return LoadUsageChunk(data, ctx) case "error": return LoadErrorChunk(data, ctx) + case "failure": + return LoadFailureChunk(data, ctx) default: return nil, fmt.Errorf("unknown StreamChunk discriminator field 'kind' value: %s", discriminator) } @@ -484,3 +486,87 @@ func ErrorChunkFromYAML(yamlStr string) (ErrorChunk, error) { ctx := NewLoadContext() return LoadErrorChunk(data, ctx) } + +// FailureChunk represents A classified failure chunk from the LLM response stream. + +type FailureChunk struct { + Kind string `json:"kind" yaml:"kind"` + Failure StreamFailure `json:"failure" yaml:"failure"` +} + +// LoadFailureChunk creates a FailureChunk from a map[string]interface{} +func LoadFailureChunk(data interface{}, ctx *LoadContext) (FailureChunk, error) { + if ctx == nil { + ctx = NewLoadContext() + } + result := FailureChunk{} + + // Load from map + if m, ok := data.(map[string]interface{}); ok { + if requiredValue, exists := m["failure"]; !exists || requiredValue == nil { + return result, fmt.Errorf("%s: missing required field", ctx.At("failure").Path) + } + if val, ok := m["kind"]; ok && val != nil { + result.Kind = string(val.(string)) + } + if val, ok := m["failure"]; ok && val != nil { + if m, ok := val.(map[string]interface{}); ok { + loaded, err := LoadStreamFailure(m, ctx.At("failure")) + if err != nil { + return result, err + } + result.Failure = loaded + } + } + } + + return result, nil +} + +// Save serializes FailureChunk to map[string]interface{} +func (obj FailureChunk) Save(ctx *SaveContext) map[string]interface{} { + result := make(map[string]interface{}) + result["kind"] = obj.Kind + + result["failure"] = obj.Failure.Save(ctx) + + return result +} + +// ToJSON serializes FailureChunk to JSON string +func (obj *FailureChunk) ToJSON() (string, error) { + ctx := NewSaveContext() + data := obj.Save(ctx) + bytes, err := json.Marshal(data) + if err != nil { + return "", err + } + return string(bytes), nil +} + +// ToYAML serializes FailureChunk to YAML string +func (obj *FailureChunk) ToYAML() (string, error) { + ctx := NewSaveContext() + data := obj.Save(ctx) + return marshalYAMLDocument(data) +} + +// FromJSON creates FailureChunk from JSON string +func FailureChunkFromJSON(jsonStr string) (FailureChunk, error) { + var data map[string]interface{} + if err := json.Unmarshal([]byte(jsonStr), &data); err != nil { + return FailureChunk{}, err + } + ctx := NewLoadContext() + return LoadFailureChunk(data, ctx) +} + +// FromYAML creates FailureChunk from YAML string +func FailureChunkFromYAML(yamlStr string) (FailureChunk, error) { + var data map[string]interface{} + if err := yaml.Unmarshal([]byte(yamlStr), &data); err != nil { + return FailureChunk{}, err + } + ctx := NewLoadContext() + return LoadFailureChunk(data, ctx) +} diff --git a/runtime/go/prompty/model/stream_failure.go b/runtime/go/prompty/model/stream_failure.go new file mode 100644 index 000000000..459fd030b --- /dev/null +++ b/runtime/go/prompty/model/stream_failure.go @@ -0,0 +1,90 @@ +// +// Code generated by Typra emitter; DO NOT EDIT. +// Group: contracts + +package prompty + +import ( + "encoding/json" + + "gopkg.in/yaml.v3" +) + +// StreamFailureOutcome represents the allowed values for StreamFailureOutcome. +type StreamFailureOutcome string + +const ( + StreamFailureOutcomeDeterminate StreamFailureOutcome = "determinate" + StreamFailureOutcomeIndeterminate StreamFailureOutcome = "indeterminate" +) + +// StreamFailure represents A classified terminal failure from an LLM response stream. + +type StreamFailure struct { + Outcome StreamFailureOutcome `json:"outcome" yaml:"outcome"` + Message string `json:"message" yaml:"message"` +} + +// LoadStreamFailure creates a StreamFailure from a map[string]interface{} +func LoadStreamFailure(data interface{}, ctx *LoadContext) (StreamFailure, error) { + result := StreamFailure{} + + // Load from map + if m, ok := data.(map[string]interface{}); ok { + if val, ok := m["outcome"]; ok && val != nil { + result.Outcome = StreamFailureOutcome(val.(string)) + } + if val, ok := m["message"]; ok && val != nil { + result.Message = string(val.(string)) + } + } + + return result, nil +} + +// Save serializes StreamFailure to map[string]interface{} +func (obj StreamFailure) Save(ctx *SaveContext) map[string]interface{} { + result := make(map[string]interface{}) + result["outcome"] = string(obj.Outcome) + result["message"] = obj.Message + + return result +} + +// ToJSON serializes StreamFailure to JSON string +func (obj *StreamFailure) ToJSON() (string, error) { + ctx := NewSaveContext() + data := obj.Save(ctx) + bytes, err := json.Marshal(data) + if err != nil { + return "", err + } + return string(bytes), nil +} + +// ToYAML serializes StreamFailure to YAML string +func (obj *StreamFailure) ToYAML() (string, error) { + ctx := NewSaveContext() + data := obj.Save(ctx) + return marshalYAMLDocument(data) +} + +// FromJSON creates StreamFailure from JSON string +func StreamFailureFromJSON(jsonStr string) (StreamFailure, error) { + var data map[string]interface{} + if err := json.Unmarshal([]byte(jsonStr), &data); err != nil { + return StreamFailure{}, err + } + ctx := NewLoadContext() + return LoadStreamFailure(data, ctx) +} + +// FromYAML creates StreamFailure from YAML string +func StreamFailureFromYAML(yamlStr string) (StreamFailure, error) { + var data map[string]interface{} + if err := yaml.Unmarshal([]byte(yamlStr), &data); err != nil { + return StreamFailure{}, err + } + ctx := NewLoadContext() + return LoadStreamFailure(data, ctx) +} diff --git a/runtime/go/prompty/model/stream_failure_test.go b/runtime/go/prompty/model/stream_failure_test.go new file mode 100644 index 000000000..1e4caad66 --- /dev/null +++ b/runtime/go/prompty/model/stream_failure_test.go @@ -0,0 +1,225 @@ +// +// Code generated by Typra emitter; DO NOT EDIT. + +package prompty_test + +import ( + "encoding/json" + "testing" + + "gopkg.in/yaml.v3" + + "prompty/model" +) + +// TestStreamFailureLoadJSON tests loading StreamFailure from JSON +func TestStreamFailureLoadJSON(t *testing.T) { + jsonData := ` +{ + "outcome": "indeterminate", + "message": "SSE stream error: connection reset" +} +` + var data map[string]interface{} + if err := json.Unmarshal([]byte(jsonData), &data); err != nil { + t.Fatalf("Failed to parse JSON: %v", err) + } + + ctx := prompty.NewLoadContext() + instance, err := prompty.LoadStreamFailure(data, ctx) + if err != nil { + t.Fatalf("Failed to load StreamFailure: %v", err) + } + if instance.Outcome != "indeterminate" { + t.Errorf(`Expected Outcome to be "indeterminate", got %v`, instance.Outcome) + } + if instance.Message != "SSE stream error: connection reset" { + t.Errorf(`Expected Message to be "SSE stream error: connection reset", got %v`, instance.Message) + } +} + +// TestStreamFailureLoadYAML tests loading StreamFailure from YAML +func TestStreamFailureLoadYAML(t *testing.T) { + yamlData := ` +outcome: indeterminate +message: "SSE stream error: connection reset" + +` + var data map[string]interface{} + if err := yaml.Unmarshal([]byte(yamlData), &data); err != nil { + t.Fatalf("Failed to parse YAML: %v", err) + } + + ctx := prompty.NewLoadContext() + instance, err := prompty.LoadStreamFailure(data, ctx) + if err != nil { + t.Fatalf("Failed to load StreamFailure: %v", err) + } + if instance.Outcome != "indeterminate" { + t.Errorf(`Expected Outcome to be "indeterminate", got %v`, instance.Outcome) + } + if instance.Message != "SSE stream error: connection reset" { + t.Errorf(`Expected Message to be "SSE stream error: connection reset", got %v`, instance.Message) + } +} + +// TestStreamFailureFromJSON tests loading StreamFailure through the generated JSON helper +func TestStreamFailureFromJSON(t *testing.T) { + jsonData := ` +{ + "outcome": "indeterminate", + "message": "SSE stream error: connection reset" +} +` + + instance, err := prompty.StreamFailureFromJSON(jsonData) + if err != nil { + t.Fatalf("Failed to load StreamFailure from JSON helper: %v", err) + } + if instance.Outcome != "indeterminate" { + t.Errorf(`Expected Outcome to be "indeterminate", got %v`, instance.Outcome) + } + if instance.Message != "SSE stream error: connection reset" { + t.Errorf(`Expected Message to be "SSE stream error: connection reset", got %v`, instance.Message) + } +} + +// TestStreamFailureFromYAML tests loading StreamFailure through the generated YAML helper +func TestStreamFailureFromYAML(t *testing.T) { + yamlData := ` +outcome: indeterminate +message: "SSE stream error: connection reset" + +` + + instance, err := prompty.StreamFailureFromYAML(yamlData) + if err != nil { + t.Fatalf("Failed to load StreamFailure from YAML helper: %v", err) + } + if instance.Outcome != "indeterminate" { + t.Errorf(`Expected Outcome to be "indeterminate", got %v`, instance.Outcome) + } + if instance.Message != "SSE stream error: connection reset" { + t.Errorf(`Expected Message to be "SSE stream error: connection reset", got %v`, instance.Message) + } +} + +// TestStreamFailureRoundtrip tests load -> save -> load produces equivalent data +func TestStreamFailureRoundtrip(t *testing.T) { + jsonData := ` +{ + "outcome": "indeterminate", + "message": "SSE stream error: connection reset" +} +` + var data map[string]interface{} + if err := json.Unmarshal([]byte(jsonData), &data); err != nil { + t.Fatalf("Failed to parse JSON: %v", err) + } + + loadCtx := prompty.NewLoadContext() + instance, err := prompty.LoadStreamFailure(data, loadCtx) + if err != nil { + t.Fatalf("Failed to load StreamFailure: %v", err) + } + saveCtx := prompty.NewSaveContext() + savedData := instance.Save(saveCtx) + + reloaded, err := prompty.LoadStreamFailure(savedData, loadCtx) + if err != nil { + t.Fatalf("Failed to reload StreamFailure: %v", err) + } + if reloaded.Outcome != "indeterminate" { + t.Errorf(`Expected Outcome to be "indeterminate", got %v`, reloaded.Outcome) + } + if reloaded.Message != "SSE stream error: connection reset" { + t.Errorf(`Expected Message to be "SSE stream error: connection reset", got %v`, reloaded.Message) + } +} + +// TestStreamFailureToJSON tests that ToJSON produces valid JSON +func TestStreamFailureToJSON(t *testing.T) { + jsonData := ` +{ + "outcome": "indeterminate", + "message": "SSE stream error: connection reset" +} +` + var data map[string]interface{} + if err := json.Unmarshal([]byte(jsonData), &data); err != nil { + t.Fatalf("Failed to parse JSON: %v", err) + } + + ctx := prompty.NewLoadContext() + instance, err := prompty.LoadStreamFailure(data, ctx) + if err != nil { + t.Fatalf("Failed to load StreamFailure: %v", err) + } + jsonOutput, err := instance.ToJSON() + if err != nil { + t.Fatalf("Failed to convert to JSON: %v", err) + } + + var parsed map[string]interface{} + if err := json.Unmarshal([]byte(jsonOutput), &parsed); err != nil { + t.Fatalf("Failed to parse generated JSON: %v", err) + } + + reloaded, err := prompty.LoadStreamFailure(parsed, ctx) + if err != nil { + t.Fatalf("Failed to reload generated JSON: %v", err) + } + if reloaded.Outcome != "indeterminate" { + t.Errorf(`Expected Outcome to be "indeterminate", got %v`, reloaded.Outcome) + } + if reloaded.Message != "SSE stream error: connection reset" { + t.Errorf(`Expected Message to be "SSE stream error: connection reset", got %v`, reloaded.Message) + } +} + +// TestStreamFailureToYAML tests that ToYAML produces valid YAML +func TestStreamFailureToYAML(t *testing.T) { + jsonData := ` +{ + "outcome": "indeterminate", + "message": "SSE stream error: connection reset" +} +` + var data map[string]interface{} + if err := json.Unmarshal([]byte(jsonData), &data); err != nil { + t.Fatalf("Failed to parse JSON: %v", err) + } + + ctx := prompty.NewLoadContext() + instance, err := prompty.LoadStreamFailure(data, ctx) + if err != nil { + t.Fatalf("Failed to load StreamFailure: %v", err) + } + yamlOutput, err := instance.ToYAML() + if err != nil { + t.Fatalf("Failed to convert to YAML: %v", err) + } + + var parsed map[string]interface{} + if err := yaml.Unmarshal([]byte(yamlOutput), &parsed); err != nil { + t.Fatalf("Failed to parse generated YAML: %v", err) + } + + reloaded, err := prompty.LoadStreamFailure(parsed, ctx) + if err != nil { + t.Fatalf("Failed to reload generated YAML: %v", err) + } + if reloaded.Outcome != "indeterminate" { + t.Errorf(`Expected Outcome to be "indeterminate", got %v`, reloaded.Outcome) + } + if reloaded.Message != "SSE stream error: connection reset" { + t.Errorf(`Expected Message to be "SSE stream error: connection reset", got %v`, reloaded.Message) + } +} + +// TestStreamFailureFromJSONInvalid rejects malformed JSON instead of silently defaulting +func TestStreamFailureFromJSONInvalid(t *testing.T) { + if _, err := prompty.StreamFailureFromJSON("{"); err == nil { + t.Fatalf("Expected malformed JSON to fail") + } +} diff --git a/runtime/python/prompty/prompty/model/__init__.py b/runtime/python/prompty/prompty/model/__init__.py index 34754c112..95199ff47 100644 --- a/runtime/python/prompty/prompty/model/__init__.py +++ b/runtime/python/prompty/prompty/model/__init__.py @@ -67,6 +67,7 @@ DoneEventPayload, ErrorChunk, ErrorEventPayload, + FailureChunk, HarnessContext, HookEndPayload, HookStartPayload, @@ -92,6 +93,7 @@ SessionWarningPayload, StatusEventPayload, StreamChunk, + StreamFailure, TextChunk, ThinkingChunk, ThinkingEventPayload, @@ -331,12 +333,14 @@ "SessionRef", "SessionSummary", "SessionTrace", + "StreamFailure", "StreamChunk", "TextChunk", "ThinkingChunk", "ToolChunk", "UsageChunk", "ErrorChunk", + "FailureChunk", "StreamOptions", "TraceTime", "TraceSpan", diff --git a/runtime/python/prompty/prompty/model/contracts/events/_StreamChunk.py b/runtime/python/prompty/prompty/model/contracts/events/_StreamChunk.py index 7c01bcf1e..473ce5339 100644 --- a/runtime/python/prompty/prompty/model/contracts/events/_StreamChunk.py +++ b/runtime/python/prompty/prompty/model/contracts/events/_StreamChunk.py @@ -12,6 +12,7 @@ from ..._context import LoadContext, SaveContext from ...contracts.conversation._ToolCall import ToolCall from ...contracts.models._InvocationUsage import InvocationUsage +from ._StreamFailure import StreamFailure @dataclass @@ -73,6 +74,8 @@ def load_kind(data: dict, context: LoadContext | None) -> "StreamChunk": return UsageChunk.load(data, context) elif discriminator_value == "error": return ErrorChunk.load(data, context) + elif discriminator_value == "failure": + return FailureChunk.load(data, context) else: raise ValueError(f"Unknown StreamChunk discriminator field 'kind' value: {discriminator_value}") @@ -591,3 +594,98 @@ def to_json(self, context: SaveContext | None = None, indent: int = 2) -> str: if context is None: context = SaveContext() return context.to_json(self.save(context), indent) + + +@dataclass +class FailureChunk(StreamChunk): + """A classified failure chunk from the LLM response stream. + + Attributes + ---------- + kind : str + The kind identifier for classified failure chunks + failure : StreamFailure + The classified stream failure + """ + + _shorthand_property: ClassVar[str | None] = None + + kind: str = field(default="failure") + failure: StreamFailure = field(default_factory=StreamFailure) + + @staticmethod + def load(data: Any, context: LoadContext | None = None) -> "FailureChunk": + """Load a FailureChunk instance. + Args: + data (Any): The data to load the instance from. + context (Optional[LoadContext]): Optional context with pre/post processing callbacks. + Returns: + FailureChunk: The loaded FailureChunk instance. + + """ + + if context is None: + context = LoadContext() + data = context.process_input(data) + + if not isinstance(data, dict): + raise ValueError(f"Invalid data for FailureChunk: {data}") + if "failure" not in data or data["failure"] is None: + raise ValueError(f"{context.at('failure').path}: missing required field") + + # create new instance + instance = FailureChunk() + + if data is not None and "kind" in data: + instance.kind = data["kind"] + if data is not None and "failure" in data: + instance.failure = StreamFailure.load(data["failure"], context.at("failure")) + if context is not None: + instance = context.process_output(instance) + return instance + + def save(self, context: SaveContext | None = None) -> dict[str, Any]: + """Save the FailureChunk instance to a dictionary. + Args: + context (Optional[SaveContext]): Optional context with pre/post processing callbacks. + Returns: + dict[str, Any]: The dictionary representation of this instance. + + """ + obj = self + if context is not None: + obj = context.process_object(obj) + + # Start with parent class properties + result = super().save(context) + + if obj.kind is not None: + result["kind"] = obj.kind + if obj.failure is not None: + result["failure"] = obj.failure.save(context) + return result + + def to_yaml(self, context: SaveContext | None = None) -> str: + """Convert the FailureChunk instance to a YAML string. + Args: + context (Optional[SaveContext]): Optional context with pre/post processing callbacks. + Returns: + str: The YAML string representation of this instance. + + """ + if context is None: + context = SaveContext() + return context.to_yaml(self.save(context)) + + def to_json(self, context: SaveContext | None = None, indent: int = 2) -> str: + """Convert the FailureChunk instance to a JSON string. + Args: + context (Optional[SaveContext]): Optional context with pre/post processing callbacks. + indent (int): Number of spaces for indentation. Defaults to 2. + Returns: + str: The JSON string representation of this instance. + + """ + if context is None: + context = SaveContext() + return context.to_json(self.save(context), indent) diff --git a/runtime/python/prompty/prompty/model/contracts/events/_StreamFailure.py b/runtime/python/prompty/prompty/model/contracts/events/_StreamFailure.py new file mode 100644 index 000000000..6ffdb54bb --- /dev/null +++ b/runtime/python/prompty/prompty/model/contracts/events/_StreamFailure.py @@ -0,0 +1,108 @@ +# +########################################## +# WARNING: This is an auto-generated file. +# DO NOT EDIT THIS FILE DIRECTLY +# ANY EDITS WILL BE LOST +########################################## + +from dataclasses import dataclass, field +from typing import Any, ClassVar, Literal + +from ..._context import LoadContext, SaveContext + +StreamFailureOutcome = Literal["determinate", "indeterminate"] + + +@dataclass +class StreamFailure: + """A classified terminal failure from an LLM response stream. + + Attributes + ---------- + outcome : str + Whether the provider outcome is known or requires reconciliation + message : str + The human-readable failure message + """ + + _shorthand_property: ClassVar[str | None] = None + + outcome: StreamFailureOutcome = field(default="determinate") + message: str = field(default="") + + @staticmethod + def load(data: Any, context: LoadContext | None = None) -> "StreamFailure": + """Load a StreamFailure instance. + Args: + data (Any): The data to load the instance from. + context (Optional[LoadContext]): Optional context with pre/post processing callbacks. + Returns: + StreamFailure: The loaded StreamFailure instance. + + """ + + if context is None: + context = LoadContext() + data = context.process_input(data) + + if not isinstance(data, dict): + raise ValueError(f"Invalid data for StreamFailure: {data}") + + # create new instance + instance = StreamFailure() + + if data is not None and "outcome" in data: + instance.outcome = data["outcome"] + if data is not None and "message" in data: + instance.message = data["message"] + if context is not None: + instance = context.process_output(instance) + return instance + + def save(self, context: SaveContext | None = None) -> dict[str, Any]: + """Save the StreamFailure instance to a dictionary. + Args: + context (Optional[SaveContext]): Optional context with pre/post processing callbacks. + Returns: + dict[str, Any]: The dictionary representation of this instance. + + """ + obj = self + if context is not None: + obj = context.process_object(obj) + + result: dict[str, Any] = {} + + if obj.outcome is not None: + result["outcome"] = obj.outcome + if obj.message is not None: + result["message"] = obj.message + + if context is not None: + result = context.process_dict(result) + return result + + def to_yaml(self, context: SaveContext | None = None) -> str: + """Convert the StreamFailure instance to a YAML string. + Args: + context (Optional[SaveContext]): Optional context with pre/post processing callbacks. + Returns: + str: The YAML string representation of this instance. + + """ + if context is None: + context = SaveContext() + return context.to_yaml(self.save(context)) + + def to_json(self, context: SaveContext | None = None, indent: int = 2) -> str: + """Convert the StreamFailure instance to a JSON string. + Args: + context (Optional[SaveContext]): Optional context with pre/post processing callbacks. + indent (int): Number of spaces for indentation. Defaults to 2. + Returns: + str: The JSON string representation of this instance. + + """ + if context is None: + context = SaveContext() + return context.to_json(self.save(context), indent) diff --git a/runtime/python/prompty/prompty/model/contracts/events/__init__.py b/runtime/python/prompty/prompty/model/contracts/events/__init__.py index d5f56fbdf..5b585f121 100644 --- a/runtime/python/prompty/prompty/model/contracts/events/__init__.py +++ b/runtime/python/prompty/prompty/model/contracts/events/__init__.py @@ -36,12 +36,14 @@ from ._StatusEventPayload import StatusEventPayload from ._StreamChunk import ( ErrorChunk, + FailureChunk, StreamChunk, TextChunk, ThinkingChunk, ToolChunk, UsageChunk, ) +from ._StreamFailure import StreamFailure from ._ThinkingEventPayload import ThinkingEventPayload from ._TokenEventPayload import TokenEventPayload from ._ToolCallCompletePayload import ToolCallCompletePayload @@ -100,10 +102,12 @@ "SessionRef", "SessionSummary", "SessionTrace", + "StreamFailure", "StreamChunk", "TextChunk", "ThinkingChunk", "ToolChunk", "UsageChunk", "ErrorChunk", + "FailureChunk", ] diff --git a/runtime/python/prompty/tests/model/contracts/events/test_failure_chunk.py b/runtime/python/prompty/tests/model/contracts/events/test_failure_chunk.py new file mode 100644 index 000000000..a13d0f481 --- /dev/null +++ b/runtime/python/prompty/tests/model/contracts/events/test_failure_chunk.py @@ -0,0 +1,85 @@ +# +import json + +import yaml + +from prompty.model import FailureChunk + + +def test_load_json_failurechunk(): + json_data = r""" + { + "failure": { + "outcome": "indeterminate", + "message": "SSE stream error: connection reset" + } + } + """ + data = json.loads(json_data, strict=False) + instance = FailureChunk.load(data) + assert instance is not None + + +def test_load_yaml_failurechunk(): + yaml_data = r""" + failure: + outcome: indeterminate + message: "SSE stream error: connection reset" + + """ + data = yaml.load(yaml_data, Loader=yaml.FullLoader) + instance = FailureChunk.load(data) + assert instance is not None + + +def test_roundtrip_json_failurechunk(): + """Test that load -> save -> load produces equivalent data.""" + json_data = r""" + { + "failure": { + "outcome": "indeterminate", + "message": "SSE stream error: connection reset" + } + } + """ + original_data = json.loads(json_data, strict=False) + instance = FailureChunk.load(original_data) + saved_data = instance.save() + reloaded = FailureChunk.load(saved_data) + assert reloaded is not None + + +def test_to_json_failurechunk(): + """Test that to_json produces valid JSON.""" + json_data = r""" + { + "failure": { + "outcome": "indeterminate", + "message": "SSE stream error: connection reset" + } + } + """ + data = json.loads(json_data, strict=False) + instance = FailureChunk.load(data) + json_output = instance.to_json() + assert json_output is not None + parsed = json.loads(json_output) + assert isinstance(parsed, dict) + + +def test_to_yaml_failurechunk(): + """Test that to_yaml produces valid YAML.""" + json_data = r""" + { + "failure": { + "outcome": "indeterminate", + "message": "SSE stream error: connection reset" + } + } + """ + data = json.loads(json_data, strict=False) + instance = FailureChunk.load(data) + yaml_output = instance.to_yaml() + assert yaml_output is not None + parsed = yaml.safe_load(yaml_output) + assert isinstance(parsed, dict) diff --git a/runtime/python/prompty/tests/model/contracts/events/test_stream_failure.py b/runtime/python/prompty/tests/model/contracts/events/test_stream_failure.py new file mode 100644 index 000000000..a611f9951 --- /dev/null +++ b/runtime/python/prompty/tests/model/contracts/events/test_stream_failure.py @@ -0,0 +1,82 @@ +# +import json + +import yaml + +from prompty.model import StreamFailure + + +def test_load_json_streamfailure(): + json_data = r""" + { + "outcome": "indeterminate", + "message": "SSE stream error: connection reset" + } + """ + data = json.loads(json_data, strict=False) + instance = StreamFailure.load(data) + assert instance is not None + assert instance.outcome == "indeterminate" + assert instance.message == "SSE stream error: connection reset" + + +def test_load_yaml_streamfailure(): + yaml_data = r""" + outcome: indeterminate + message: "SSE stream error: connection reset" + + """ + data = yaml.load(yaml_data, Loader=yaml.FullLoader) + instance = StreamFailure.load(data) + assert instance is not None + assert instance.outcome == "indeterminate" + assert instance.message == "SSE stream error: connection reset" + + +def test_roundtrip_json_streamfailure(): + """Test that load -> save -> load produces equivalent data.""" + json_data = r""" + { + "outcome": "indeterminate", + "message": "SSE stream error: connection reset" + } + """ + original_data = json.loads(json_data, strict=False) + instance = StreamFailure.load(original_data) + saved_data = instance.save() + reloaded = StreamFailure.load(saved_data) + assert reloaded is not None + assert reloaded.outcome == "indeterminate" + assert reloaded.message == "SSE stream error: connection reset" + + +def test_to_json_streamfailure(): + """Test that to_json produces valid JSON.""" + json_data = r""" + { + "outcome": "indeterminate", + "message": "SSE stream error: connection reset" + } + """ + data = json.loads(json_data, strict=False) + instance = StreamFailure.load(data) + json_output = instance.to_json() + assert json_output is not None + parsed = json.loads(json_output) + assert isinstance(parsed, dict) + + +def test_to_yaml_streamfailure(): + """Test that to_yaml produces valid YAML.""" + json_data = r""" + { + "outcome": "indeterminate", + "message": "SSE stream error: connection reset" + } + """ + data = json.loads(json_data, strict=False) + instance = StreamFailure.load(data) + yaml_output = instance.to_yaml() + assert yaml_output is not None + parsed = yaml.safe_load(yaml_output) + assert isinstance(parsed, dict) diff --git a/runtime/rust/prompty-openai/src/processor.rs b/runtime/rust/prompty-openai/src/processor.rs index fa807d10e..597ecd479 100644 --- a/runtime/rust/prompty-openai/src/processor.rs +++ b/runtime/rust/prompty-openai/src/processor.rs @@ -617,8 +617,10 @@ impl futures::Stream for OpenAIStreamProcessor { if let Some(refusal) = chunk.get("delta").and_then(Value::as_str) { if !refusal.is_empty() { this.phase = StreamPhase::Done; - return std::task::Poll::Ready(Some(StreamChunk::Error( - format!("Model refused: {refusal}"), + return std::task::Poll::Ready(Some(StreamChunk::Failure( + StreamFailure::Determinate(format!( + "Model refused: {refusal}" + )), ))); } } @@ -679,8 +681,10 @@ impl futures::Stream for OpenAIStreamProcessor { if let Some(refusal) = delta.get("refusal").and_then(Value::as_str) { if !refusal.is_empty() { this.phase = StreamPhase::Done; - return std::task::Poll::Ready(Some(StreamChunk::Error( - format!("Model refused: {refusal}"), + return std::task::Poll::Ready(Some(StreamChunk::Failure( + StreamFailure::Determinate(format!( + "Model refused: {refusal}" + )), ))); } } @@ -1299,14 +1303,14 @@ mod tests { let chunks = vec![json!({"choices": [{"delta": {"refusal": "I cannot help with that"}}]})]; let inner = futures::stream::iter(chunks); let mut stream = process_stream(inner); - let mut errors = Vec::new(); + let mut failures = Vec::new(); while let Some(chunk) = stream.next().await { - if let StreamChunk::Error(message) = chunk { - errors.push(message); + if let StreamChunk::Failure(StreamFailure::Determinate(message)) = chunk { + failures.push(message); } } - assert_eq!(errors.len(), 1); - assert!(errors[0].contains("refused")); + assert_eq!(failures.len(), 1); + assert!(failures[0].contains("refused")); } #[tokio::test] diff --git a/runtime/rust/prompty-openai/tests/stream_failure_vectors.rs b/runtime/rust/prompty-openai/tests/stream_failure_vectors.rs new file mode 100644 index 000000000..4576bafed --- /dev/null +++ b/runtime/rust/prompty-openai/tests/stream_failure_vectors.rs @@ -0,0 +1,83 @@ +use futures::StreamExt; +use prompty::types::{StreamChunk, StreamFailure}; +use serde_json::{Value, json}; + +fn load_vectors() -> Vec { + let path = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .join("..") + .join("..") + .join("..") + .join("spec") + .join("vectors") + .join("process") + .join("stream_failure_vectors.json"); + serde_json::from_str( + &std::fs::read_to_string(&path) + .unwrap_or_else(|error| panic!("Failed to read {}: {error}", path.display())), + ) + .expect("Invalid stream failure vectors") +} + +fn provider_chunks(events: &[Value]) -> Vec { + events + .iter() + .map(|event| match event["kind"].as_str() { + Some("provider") => event["value"].clone(), + Some("transportError") => json!({ + "error": { + "type": "sse_transport_error", + "message": event["message"], + } + }), + kind => panic!("Unsupported stream vector event kind: {kind:?}"), + }) + .collect() +} + +fn chunk_to_value(chunk: StreamChunk) -> Value { + match chunk { + StreamChunk::Text(value) => json!({"kind": "text", "value": value}), + StreamChunk::Failure(failure) => json!({ + "kind": "failure", + "failure": { + "outcome": if failure.outcome_unknown() { + "indeterminate" + } else { + "determinate" + }, + "message": failure.message(), + } + }), + other => panic!("Unexpected processed stream chunk: {other:?}"), + } +} + +#[tokio::test] +async fn openai_stream_processor_matches_classified_failure_vectors() { + for vector in load_vectors() { + let input = &vector["input"]; + assert_eq!(input["provider"], "openai"); + let events = input["events"] + .as_array() + .expect("stream vector events must be an array"); + let chunks = provider_chunks(events); + let actual: Vec = + prompty_openai::processor::process_stream(futures::stream::iter(chunks)) + .map(chunk_to_value) + .collect() + .await; + + assert_eq!( + Value::Array(actual), + vector["expected"]["chunks"], + "stream failure vector '{}' did not match", + vector["name"] + ); + } +} + +#[test] +fn compatibility_failure_outcomes_match_vector_contract() { + assert!(!StreamFailure::Determinate(String::new()).outcome_unknown()); + assert!(StreamFailure::Indeterminate(String::new()).outcome_unknown()); +} diff --git a/runtime/rust/prompty/src/model/contracts/events/mod.rs b/runtime/rust/prompty/src/model/contracts/events/mod.rs index cd8ecd9ff..ece9895c8 100644 --- a/runtime/rust/prompty/src/model/contracts/events/mod.rs +++ b/runtime/rust/prompty/src/model/contracts/events/mod.rs @@ -103,6 +103,9 @@ pub use status_event_payload::*; pub mod stream_chunk; pub use stream_chunk::*; +pub mod stream_failure; +pub use stream_failure::*; + pub mod thinking_event_payload; pub use thinking_event_payload::*; diff --git a/runtime/rust/prompty/src/model/contracts/events/stream_chunk.rs b/runtime/rust/prompty/src/model/contracts/events/stream_chunk.rs index f295d85eb..f6e311d28 100644 --- a/runtime/rust/prompty/src/model/contracts/events/stream_chunk.rs +++ b/runtime/rust/prompty/src/model/contracts/events/stream_chunk.rs @@ -14,6 +14,8 @@ use super::super::super::context::{LoadContext, SaveContext}; use super::super::super::contracts::models::invocation_usage::InvocationUsage; +use super::stream_failure::StreamFailure; + use super::super::super::contracts::conversation::tool_call::ToolCall; /// Variant-specific data for [`StreamChunk`], discriminated by `kind`. @@ -44,6 +46,11 @@ pub enum StreamChunkKind { /// The error message message: String, }, + /// `kind` = `"failure"` + FailureChunk { + /// The classified stream failure + failure: StreamFailure, + }, } impl Default for StreamChunkKind { @@ -127,6 +134,13 @@ impl StreamChunk { .unwrap_or_default() .to_string(), }, + "failure" => StreamChunkKind::FailureChunk { + failure: value + .get("failure") + .filter(|v| v.is_object() || v.is_array() || v.is_string()) + .map(|v| StreamFailure::load_from_value(v, ctx)) + .unwrap_or_default(), + }, _ => panic!( "Unknown StreamChunk discriminator field 'kind' value: {}", kind_str @@ -177,6 +191,18 @@ impl StreamChunk { InvocationUsage::validate_input_at(child, &child_path)?; } "error" => {} + "failure" => { + let child_path = if path.is_empty() { + "failure".to_string() + } else { + format!("{}.failure", path) + }; + let child = value + .get("failure") + .filter(|candidate| !candidate.is_null()) + .ok_or_else(|| format!("{}: missing required field", child_path))?; + StreamFailure::validate_input_at(child, &child_path)?; + } _ => {} } Ok(()) @@ -188,7 +214,7 @@ impl StreamChunk { .and_then(|candidate| candidate.as_str()) .ok_or_else(|| "Missing StreamChunk discriminator property: 'kind'".to_string())?; match discriminator { - "text" | "thinking" | "tool" | "usage" | "error" => Ok(()), + "text" | "thinking" | "tool" | "usage" | "error" | "failure" => Ok(()), _ => Err(format!( "Unknown StreamChunk discriminator field 'kind' value: {}", discriminator @@ -204,6 +230,7 @@ impl StreamChunk { StreamChunkKind::ToolChunk { .. } => "tool", StreamChunkKind::UsageChunk { .. } => "usage", StreamChunkKind::ErrorChunk { .. } => "error", + StreamChunkKind::FailureChunk { .. } => "failure", } } @@ -246,6 +273,10 @@ impl StreamChunk { serde_json::Value::String(message.clone()), ); } + StreamChunkKind::FailureChunk { failure, .. } => { + let nested = failure.to_value(ctx); + result.insert("failure".to_string(), nested); + } } ctx.process_dict(serde_json::Value::Object(result)) } diff --git a/runtime/rust/prompty/src/model/contracts/events/stream_failure.rs b/runtime/rust/prompty/src/model/contracts/events/stream_failure.rs new file mode 100644 index 000000000..7f07d4056 --- /dev/null +++ b/runtime/rust/prompty/src/model/contracts/events/stream_failure.rs @@ -0,0 +1,182 @@ +// +// Code generated by Typra emitter; DO NOT EDIT. + +#![allow( + unused_imports, + dead_code, + non_camel_case_types, + unused_variables, + unexpected_cfgs, + clippy::all +)] + +use super::super::super::context::{LoadContext, SaveContext}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum StreamFailureOutcome { + Determinate, + Indeterminate, +} + +impl Default for StreamFailureOutcome { + fn default() -> Self { + Self::Determinate + } +} + +impl std::fmt::Display for StreamFailureOutcome { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Determinate => write!(f, "determinate"), + Self::Indeterminate => write!(f, "indeterminate"), + } + } +} + +impl StreamFailureOutcome { + pub fn from_str_opt(s: &str) -> Option { + match s { + "determinate" => Some(Self::Determinate), + "indeterminate" => Some(Self::Indeterminate), + _ => None, + } + } + + pub fn from_str_ignore_case_opt(s: &str) -> Option { + if s.eq_ignore_ascii_case("determinate") { + return Some(Self::Determinate); + } + if s.eq_ignore_ascii_case("indeterminate") { + return Some(Self::Indeterminate); + } + None + } + + pub fn as_str(&self) -> &str { + match self { + Self::Determinate => "determinate", + Self::Indeterminate => "indeterminate", + } + } +} + +#[cfg(feature = "serde")] +impl serde::Serialize for StreamFailureOutcome { + fn serialize(&self, serializer: S) -> Result { + serializer.serialize_str(self.as_str()) + } +} + +#[cfg(feature = "serde")] +impl<'de> serde::Deserialize<'de> for StreamFailureOutcome { + fn deserialize>(deserializer: D) -> Result { + let s = ::deserialize(deserializer)?; + Self::from_str_opt(&s).ok_or_else(|| { + serde::de::Error::custom(format!("invalid StreamFailureOutcome value: {}", s)) + }) + } +} + +/// A classified terminal failure from an LLM response stream. +#[derive(Debug, Clone, Default, PartialEq)] +pub struct StreamFailure { + /// Whether the provider outcome is known or requires reconciliation + pub outcome: StreamFailureOutcome, + /// The human-readable failure message + pub message: String, +} + +impl StreamFailure { + /// Create a new StreamFailure with default values. + pub fn new() -> Self { + Self::default() + } + + /// Load StreamFailure from a JSON string. + pub fn from_json(json: &str, ctx: &LoadContext) -> Result { + let value: serde_json::Value = serde_json::from_str(json)?; + Self::validate_input_at(&value, "") + .map_err(|message| ::custom(message))?; + Ok(Self::load_from_value(&value, ctx)) + } + + /// Load StreamFailure from a YAML string. + pub fn from_yaml(yaml: &str, ctx: &LoadContext) -> Result { + let value: serde_json::Value = serde_yaml::from_str(yaml)?; + Self::validate_input_at(&value, "") + .map_err(|message| ::custom(message))?; + Ok(Self::load_from_value(&value, ctx)) + } + + /// Load StreamFailure from a `serde_json::Value`. + /// + /// Calls `ctx.process_input` before field extraction. + pub fn load_from_value(value: &serde_json::Value, ctx: &LoadContext) -> Self { + let value = ctx.process_input(value.clone()); + if let Err(message) = Self::validate_input_at(&value, "") { + panic!("{}", message); + } + Self { + outcome: value + .get("outcome") + .and_then(|v| v.as_str()) + .and_then(|s| StreamFailureOutcome::from_str_opt(s)) + .unwrap_or(StreamFailureOutcome::Determinate), + message: value + .get("message") + .and_then(|v| v.as_str()) + .unwrap_or_default() + .to_string(), + } + } + + pub(crate) fn validate_input_at(value: &serde_json::Value, path: &str) -> Result<(), String> { + Ok(()) + } + + /// Serialize StreamFailure to a `serde_json::Value`. + /// + /// Calls `ctx.process_dict` after serialization. + pub fn to_value(&self, ctx: &SaveContext) -> serde_json::Value { + let mut result = serde_json::Map::new(); + // Write base fields + result.insert( + "outcome".to_string(), + serde_json::Value::String(self.outcome.to_string()), + ); + result.insert( + "message".to_string(), + serde_json::Value::String(self.message.clone()), + ); + ctx.process_dict(serde_json::Value::Object(result)) + } + + /// Serialize StreamFailure to a JSON string. + pub fn to_json(&self, ctx: &SaveContext) -> Result { + serde_json::to_string_pretty(&self.to_value(ctx)) + } + + /// Serialize StreamFailure to a YAML string. + pub fn to_yaml(&self, ctx: &SaveContext) -> Result { + serde_yaml::to_string(&self.to_value(ctx)) + } +} + +// Serde for `StreamFailure` delegates to the canonical to_value/load_from_value +// logic so its serde wire form always equals the canonical to_value/load_from_value form. Uses a default (no-op) context — no ${env:}/${file:} +// resolution here — leaving the context-aware LoadContext/SaveContext API intact. +#[cfg(feature = "serde")] +impl serde::Serialize for StreamFailure { + fn serialize(&self, serializer: S) -> Result { + serde::Serialize::serialize(&self.to_value(&SaveContext::default()), serializer) + } +} + +#[cfg(feature = "serde")] +impl<'de> serde::Deserialize<'de> for StreamFailure { + fn deserialize>(deserializer: D) -> Result { + let value = ::deserialize(deserializer)?; + Self::validate_input_at(&value, "").map_err(serde::de::Error::custom)?; + Ok(Self::load_from_value(&value, &LoadContext::default())) + } +} diff --git a/runtime/rust/prompty/src/pipeline/live_turn.rs b/runtime/rust/prompty/src/pipeline/live_turn.rs index 7f1dcbfe6..0c430e8d7 100644 --- a/runtime/rust/prompty/src/pipeline/live_turn.rs +++ b/runtime/rust/prompty/src/pipeline/live_turn.rs @@ -1430,9 +1430,30 @@ mod tests { use crate::types::StreamFailure; const RESPONSES_STREAM_PROVIDER: &str = "live-responses-stream-test"; + const DETERMINATE_STREAM_PROVIDER: &str = "live-determinate-stream-test"; const INDETERMINATE_STREAM_PROVIDER: &str = "live-indeterminate-stream-test"; const POST_OPEN_INDETERMINATE_STREAM_PROVIDER: &str = "live-post-open-indeterminate-stream-test"; + + fn stream_failure_vector(name: &str) -> Value { + let path = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .join("..") + .join("..") + .join("..") + .join("spec") + .join("vectors") + .join("process") + .join("stream_failure_vectors.json"); + let vectors: Vec = serde_json::from_str( + &std::fs::read_to_string(&path) + .unwrap_or_else(|error| panic!("Failed to read {}: {error}", path.display())), + ) + .expect("Invalid stream failure vectors"); + vectors + .into_iter() + .find(|vector| vector["name"] == name) + .unwrap_or_else(|| panic!("Missing stream failure vector '{name}'")) + } static INDETERMINATE_STREAM_OPEN_CALLS: AtomicU64 = AtomicU64::new(0); static INDETERMINATE_NON_STREAM_CALLS: AtomicU64 = AtomicU64::new(0); @@ -1892,12 +1913,38 @@ mod tests { struct ResponsesStreamProcessor; + struct DeterminateStreamExecutor; + + struct DeterminateStreamProcessor; + struct IndeterminateStreamExecutor; struct PostOpenIndeterminateStreamExecutor; struct PostOpenIndeterminateStreamProcessor; + #[async_trait] + impl Executor for DeterminateStreamExecutor { + async fn execute( + &self, + _agent: &Agent, + _messages: &[Message], + ) -> Result { + unreachable!("the test exercises the streaming path") + } + + async fn execute_stream_with_context( + &self, + _agent: &Agent, + _request: &ModelInvocationRequest, + _cancellation: &CancellationToken, + ) -> Result + Send>>, InvokerError> { + Ok(Box::pin(futures::stream::iter(vec![json!({ + "refusal": "I cannot help with that" + })]))) + } + } + #[async_trait] impl Executor for IndeterminateStreamExecutor { async fn execute( @@ -2049,6 +2096,22 @@ mod tests { } } + #[async_trait] + impl Processor for DeterminateStreamProcessor { + async fn process(&self, _agent: &Agent, _response: Value) -> Result { + unreachable!("the test exercises the streaming path") + } + + fn process_stream( + &self, + _stream: Pin + Send>>, + ) -> Result + Send>>, InvokerError> { + Ok(Box::pin(futures::stream::iter(vec![StreamChunk::Failure( + StreamFailure::Determinate("Model refused: I cannot help with that".to_string()), + )]))) + } + } + fn responses_agent() -> Agent { Agent::load_from_value( &json!({ @@ -2167,6 +2230,8 @@ mod tests { #[tokio::test] async fn post_open_indeterminate_stream_requires_reconciliation_without_success_commit() { + let vector = stream_failure_vector("partial_text_then_indeterminate_failure"); + let expected = &vector["expected"]; registry::register_executor( POST_OPEN_INDETERMINATE_STREAM_PROVIDER, PostOpenIndeterminateStreamExecutor, @@ -2204,6 +2269,7 @@ mod tests { Some( TurnOptions::builder() .durability(durability.clone()) + .max_llm_retries(1) .on_event(Box::new(move |event| { captured_events .lock() @@ -2218,15 +2284,19 @@ mod tests { assert!(matches!(error, InvokerError::ExecuteIndeterminate { .. })); let events = events.lock().expect("event lock poisoned"); - assert!( - events - .iter() - .any(|event| { matches!(event, AgentEvent::Token(value) if value == "partial") }) - ); - assert!( - !events - .iter() - .any(|event| matches!(event, AgentEvent::Done { .. })) + assert!(events.iter().any(|event| { + matches!( + event, + AgentEvent::Token(value) + if Some(value.as_str()) == expected["partialText"].as_str() + ) + })); + let completion_committed = events + .iter() + .any(|event| matches!(event, AgentEvent::Done { .. })); + assert_eq!( + completion_committed, + expected["completionCommitted"].as_bool().unwrap() ); drop(events); let checkpoint = durability @@ -2236,11 +2306,82 @@ mod tests { .last() .cloned() .expect("indeterminate invocation must persist a reconciliation checkpoint"); - assert!(checkpoint.reconciliation_required); + assert_eq!( + checkpoint.reconciliation_required, + expected["requiresReconciliation"].as_bool().unwrap() + ); assert!(!checkpoint.final_output_ready); assert!(checkpoint.pending_output.is_none()); } + #[tokio::test] + async fn determinate_stream_failure_does_not_reconcile_or_commit_completion() { + let vector = stream_failure_vector("stream_refusal_is_determinate"); + let expected = &vector["expected"]; + registry::register_executor(DETERMINATE_STREAM_PROVIDER, DeterminateStreamExecutor); + registry::register_processor(DETERMINATE_STREAM_PROVIDER, DeterminateStreamProcessor); + let durability = Arc::new(CheckpointRecorder::default()); + let events = Arc::new(Mutex::new(Vec::new())); + let captured_events = events.clone(); + + let error = turn_with_engine_request( + &Agent::load_from_value( + &json!({ + "kind": "prompt", + "name": "determinate-stream-failure", + "instructions": "test", + "model": { + "id": "gpt-test", + "provider": DETERMINATE_STREAM_PROVIDER, + "options": { + "additionalProperties": {"stream": true} + } + } + }), + &LoadContext::default(), + ), + TurnEngineRequest::new( + "determinate-stream-session", + "determinate-stream-turn", + vec![Message::with_text(crate::types::Role::User, "hello")], + ), + Some( + TurnOptions::builder() + .durability(durability.clone()) + .max_llm_retries(1) + .on_event(Box::new(move |event| { + captured_events + .lock() + .expect("event lock poisoned") + .push(event); + })) + .build(), + ), + ) + .await + .expect_err("determinate stream failure must fail the turn"); + + assert!(!matches!(error, InvokerError::ExecuteIndeterminate { .. })); + let completion_committed = events + .lock() + .expect("event lock poisoned") + .iter() + .any(|event| matches!(event, AgentEvent::Done { .. })); + assert_eq!( + completion_committed, + expected["completionCommitted"].as_bool().unwrap() + ); + assert_eq!( + durability + .checkpoints + .lock() + .unwrap() + .iter() + .any(|checkpoint| checkpoint.reconciliation_required), + expected["requiresReconciliation"].as_bool().unwrap() + ); + } + #[tokio::test] async fn indeterminate_stream_open_requires_reconciliation_without_fallback() { INDETERMINATE_STREAM_OPEN_CALLS.store(0, Ordering::SeqCst); diff --git a/runtime/rust/prompty/src/types.rs b/runtime/rust/prompty/src/types.rs index cc6ca8bc2..0aa39b214 100644 --- a/runtime/rust/prompty/src/types.rs +++ b/runtime/rust/prompty/src/types.rs @@ -299,7 +299,7 @@ pub enum StreamChunk { /// Transport failures after a response has opened are indeterminate: the provider /// may have accepted or completed the invocation even though the local stream /// ended before a terminal response was received. -#[derive(Debug, Clone)] +#[derive(Debug, Clone, PartialEq, Eq)] pub enum StreamFailure { /// The stream failed deterministically (for example, a model refusal). Determinate(String), @@ -319,6 +319,47 @@ impl StreamFailure { pub fn outcome_unknown(&self) -> bool { matches!(self, Self::Indeterminate(_)) } + + /// Convert this compatibility type to the canonical generated model. + pub fn to_model(&self) -> crate::model::StreamFailure { + self.into() + } + + /// Convert a canonical generated model into the compatibility type. + pub fn from_model(failure: crate::model::StreamFailure) -> Self { + failure.into() + } +} + +impl From<&StreamFailure> for crate::model::StreamFailure { + fn from(failure: &StreamFailure) -> Self { + let outcome = if failure.outcome_unknown() { + crate::model::StreamFailureOutcome::Indeterminate + } else { + crate::model::StreamFailureOutcome::Determinate + }; + Self { + outcome, + message: failure.message().to_string(), + } + } +} + +impl From for crate::model::StreamFailure { + fn from(failure: StreamFailure) -> Self { + (&failure).into() + } +} + +impl From for StreamFailure { + fn from(failure: crate::model::StreamFailure) -> Self { + match failure.outcome { + crate::model::StreamFailureOutcome::Determinate => Self::Determinate(failure.message), + crate::model::StreamFailureOutcome::Indeterminate => { + Self::Indeterminate(failure.message) + } + } + } } // --------------------------------------------------------------------------- diff --git a/runtime/rust/prompty/tests/model/contracts/events/mod.rs b/runtime/rust/prompty/tests/model/contracts/events/mod.rs index c90f1b001..4c9037404 100644 --- a/runtime/rust/prompty/tests/model/contracts/events/mod.rs +++ b/runtime/rust/prompty/tests/model/contracts/events/mod.rs @@ -41,6 +41,7 @@ mod session_trace_test; mod session_warning_payload_test; mod status_event_payload_test; mod stream_chunk_test; +mod stream_failure_test; mod thinking_event_payload_test; mod token_event_payload_test; mod tool_call_complete_payload_test; diff --git a/runtime/rust/prompty/tests/model/contracts/events/stream_failure_test.rs b/runtime/rust/prompty/tests/model/contracts/events/stream_failure_test.rs new file mode 100644 index 000000000..ad4ba1fc6 --- /dev/null +++ b/runtime/rust/prompty/tests/model/contracts/events/stream_failure_test.rs @@ -0,0 +1,107 @@ +// +// Code generated by Typra emitter; DO NOT EDIT. + +#![allow( + unused_imports, + dead_code, + non_camel_case_types, + unused_variables, + unexpected_cfgs, + clippy::all +)] + +use prompty::model::StreamFailure; +use prompty::model::StreamFailureOutcome; +use prompty::model::context::{LoadContext, SaveContext}; + +#[test] +fn test_stream_failure_load_json() { + let json = r####" +{ + "outcome": "indeterminate", + "message": "SSE stream error: connection reset" +} +"####; + let ctx = LoadContext::default(); + let result = StreamFailure::from_json(json, &ctx); + assert!( + result.is_ok(), + "Failed to load from JSON: {:?}", + result.err() + ); + let instance = result.unwrap(); + assert_eq!(instance.outcome, StreamFailureOutcome::Indeterminate); + assert_eq!(instance.message, "SSE stream error: connection reset"); +} + +#[test] +fn test_stream_failure_load_yaml() { + let yaml = r####" +outcome: indeterminate +message: "SSE stream error: connection reset" + +"####; + let ctx = LoadContext::default(); + let result = StreamFailure::from_yaml(yaml, &ctx); + assert!( + result.is_ok(), + "Failed to load from YAML: {:?}", + result.err() + ); + let instance = result.unwrap(); + assert_eq!(instance.outcome, StreamFailureOutcome::Indeterminate); + assert_eq!(instance.message, "SSE stream error: connection reset"); +} + +#[test] +fn test_stream_failure_roundtrip() { + let json = r####" +{ + "outcome": "indeterminate", + "message": "SSE stream error: connection reset" +} +"####; + let load_ctx = LoadContext::default(); + let result = StreamFailure::from_json(json, &load_ctx); + assert!(result.is_ok(), "Failed to load: {:?}", result.err()); + let instance = result.unwrap(); + let save_ctx = SaveContext::default(); + let json_output = instance.to_json(&save_ctx); + assert!( + json_output.is_ok(), + "Failed to serialize to JSON: {:?}", + json_output.err() + ); +} + +#[cfg(feature = "serde")] +#[test] +fn test_stream_failure_serde_roundtrip() { + let json = r####" +{ + "outcome": "indeterminate", + "message": "SSE stream error: connection reset" +} +"####; + let instance: StreamFailure = + serde_json::from_str(json).expect("serde should deserialize canonical JSON"); + let value = serde_json::to_value(&instance).expect("serde should serialize"); + let canonical: serde_json::Value = serde_json::from_str(json).expect("canonical json parses"); + assert_eq!( + value, + instance.to_value(&SaveContext::default()), + "serde serialize must equal canonical to_value" + ); + assert_eq!( + instance, + StreamFailure::load_from_value(&canonical, &LoadContext::default()), + "serde deserialize must equal canonical load_from_value" + ); + assert_eq!( + value, canonical, + "serde must serialize to byte-identical canonical wire (empty-omission preserved; no plain-derive divergence)" + ); + let reparsed: StreamFailure = + serde_json::from_value(value).expect("serde should re-deserialize"); + assert_eq!(instance, reparsed, "serde round-trip must be stable"); +} diff --git a/runtime/rust/prompty/tests/stream_chunk_compatibility.rs b/runtime/rust/prompty/tests/stream_chunk_compatibility.rs index 9303e2fd7..5d202727b 100644 --- a/runtime/rust/prompty/tests/stream_chunk_compatibility.rs +++ b/runtime/rust/prompty/tests/stream_chunk_compatibility.rs @@ -1,3 +1,7 @@ +use prompty::model::{ + LoadContext, SaveContext, StreamChunk as ModelStreamChunk, StreamChunkKind, + StreamFailure as ModelStreamFailure, StreamFailureOutcome, +}; use prompty::{StreamChunk, StreamFailure}; fn legacy_external_processor_error() -> StreamChunk { @@ -21,3 +25,31 @@ fn classified_failures_remain_available_for_indeterminate_reconciliation() { StreamChunk::Failure(failure) if failure.outcome_unknown() )); } + +#[test] +fn classified_failures_bridge_to_the_canonical_generated_model() { + let compatibility = StreamFailure::Indeterminate("connection reset".to_string()); + let canonical = compatibility.to_model(); + + assert_eq!(canonical.outcome, StreamFailureOutcome::Indeterminate); + assert_eq!(canonical.message, "connection reset"); + assert_eq!(StreamFailure::from_model(canonical.clone()), compatibility); + + let chunk = ModelStreamChunk { + kind: StreamChunkKind::FailureChunk { failure: canonical }, + }; + let saved = chunk.to_value(&SaveContext::default()); + let loaded = ModelStreamChunk::load_from_value(&saved, &LoadContext::default()); + assert_eq!(loaded, chunk); + + let StreamChunkKind::FailureChunk { failure } = loaded.kind else { + panic!("canonical failure chunk must retain its discriminator"); + }; + assert_eq!( + StreamFailure::from_model(ModelStreamFailure { + outcome: failure.outcome, + message: failure.message, + }), + compatibility + ); +} diff --git a/runtime/swift/prompty-model/Sources/PromptyModel/Contracts/Events/stream_chunk.swift b/runtime/swift/prompty-model/Sources/PromptyModel/Contracts/Events/stream_chunk.swift index 3a9fee156..7663b976f 100644 --- a/runtime/swift/prompty-model/Sources/PromptyModel/Contracts/Events/stream_chunk.swift +++ b/runtime/swift/prompty-model/Sources/PromptyModel/Contracts/Events/stream_chunk.swift @@ -9,6 +9,7 @@ public enum StreamChunk: TypraModel { case toolChunk(ToolChunk) case usageChunk(UsageChunk) case errorChunk(ErrorChunk) + case failureChunk(FailureChunk) public static func load(_ data: Any, context: LoadContext = LoadContext()) throws -> StreamChunk { let normalizedData: Any = data @@ -23,6 +24,7 @@ public enum StreamChunk: TypraModel { case "tool": return .toolChunk(try ToolChunk.load(normalizedData, context: context)) case "usage": return .usageChunk(try UsageChunk.load(normalizedData, context: context)) case "error": return .errorChunk(try ErrorChunk.load(normalizedData, context: context)) + case "failure": return .failureChunk(try FailureChunk.load(normalizedData, context: context)) default: throw TypraRuntimeError.unknownDiscriminator(type: "StreamChunk", field: "kind", value: discriminator) } } @@ -34,6 +36,7 @@ public enum StreamChunk: TypraModel { case .toolChunk(let value): return try value.save(context) case .usageChunk(let value): return try value.save(context) case .errorChunk(let value): return try value.save(context) + case .failureChunk(let value): return try value.save(context) } } @@ -309,3 +312,56 @@ public struct ErrorChunk: TypraModel { return try TypraRuntime.yamlString(from: save(context)) } } + +/// A classified failure chunk from the LLM response stream. +public struct FailureChunk: TypraModel { + public static let shorthandProperty: String? = nil + public var kind: String = "failure" + public var failure: StreamFailure = StreamFailure() + + public init(kind: String = "failure", failure: StreamFailure = StreamFailure()) { + self.kind = kind + self.failure = failure + } + + public static func load(_ data: Any, context: LoadContext = LoadContext()) throws -> FailureChunk { + let object = try TypraRuntime.object(data, typeName: "FailureChunk") + var instance = FailureChunk() + if let value = object["kind"] { + instance.kind = try TypraRuntime.string(value, field: "kind") + } + else { + instance.kind = "failure" + } + if object["failure"] == nil || object["failure"] is NSNull { + throw TypraRuntimeError.unsupported(context.at("failure").path + ": missing required field") + } + if let value = object["failure"] { + instance.failure = try StreamFailure.load(value, context: context.at("failure")) + } + return instance + } + + public func save(_ context: SaveContext = SaveContext()) throws -> [String: Any] { + var result: [String: Any] = [:] + result["kind"] = self.kind + result["failure"] = try self.failure.save(context) + return result + } + + public static func fromJSON(_ json: String, context: LoadContext = LoadContext()) throws -> FailureChunk { + return try load(TypraRuntime.jsonObject(from: json, typeName: "FailureChunk"), context: context) + } + + public func toJSON(_ context: SaveContext = SaveContext()) throws -> String { + return try TypraRuntime.jsonString(from: save(context)) + } + + public static func fromYAML(_ yaml: String, context: LoadContext = LoadContext()) throws -> FailureChunk { + return try load(TypraRuntime.yamlObject(from: yaml, typeName: "FailureChunk"), context: context) + } + + public func toYAML(_ context: SaveContext = SaveContext()) throws -> String { + return try TypraRuntime.yamlString(from: save(context)) + } +} diff --git a/runtime/swift/prompty-model/Sources/PromptyModel/Contracts/Events/stream_failure.swift b/runtime/swift/prompty-model/Sources/PromptyModel/Contracts/Events/stream_failure.swift new file mode 100644 index 000000000..0e634a3bc --- /dev/null +++ b/runtime/swift/prompty-model/Sources/PromptyModel/Contracts/Events/stream_failure.swift @@ -0,0 +1,63 @@ +// +// Code generated by Typra emitter; DO NOT EDIT. + +import Foundation + +public enum StreamFailureOutcome: String, Codable, CaseIterable { + case determinate = "determinate" + case indeterminate = "indeterminate" + public static func parse(_ value: String) throws -> StreamFailureOutcome { + switch value { + case "determinate": return .determinate + case "indeterminate": return .indeterminate + default: throw TypraRuntimeError.invalidEnum(type: "StreamFailureOutcome", value: value) + } + } +} + +/// A classified terminal failure from an LLM response stream. +public struct StreamFailure: TypraModel { + public static let shorthandProperty: String? = nil + public var outcome: StreamFailureOutcome = (try! StreamFailureOutcome.parse("determinate")) + public var message: String = "" + + public init(outcome: StreamFailureOutcome = (try! StreamFailureOutcome.parse("determinate")), message: String = "") { + self.outcome = outcome + self.message = message + } + + public static func load(_ data: Any, context: LoadContext = LoadContext()) throws -> StreamFailure { + let object = try TypraRuntime.object(data, typeName: "StreamFailure") + var instance = StreamFailure() + if let value = object["outcome"] { + instance.outcome = try StreamFailureOutcome.parse(try TypraRuntime.string(value, field: "outcome")) + } + if let value = object["message"] { + instance.message = try TypraRuntime.string(value, field: "message") + } + return instance + } + + public func save(_ context: SaveContext = SaveContext()) throws -> [String: Any] { + var result: [String: Any] = [:] + result["outcome"] = self.outcome.rawValue + result["message"] = self.message + return result + } + + public static func fromJSON(_ json: String, context: LoadContext = LoadContext()) throws -> StreamFailure { + return try load(TypraRuntime.jsonObject(from: json, typeName: "StreamFailure"), context: context) + } + + public func toJSON(_ context: SaveContext = SaveContext()) throws -> String { + return try TypraRuntime.jsonString(from: save(context)) + } + + public static func fromYAML(_ yaml: String, context: LoadContext = LoadContext()) throws -> StreamFailure { + return try load(TypraRuntime.yamlObject(from: yaml, typeName: "StreamFailure"), context: context) + } + + public func toYAML(_ context: SaveContext = SaveContext()) throws -> String { + return try TypraRuntime.yamlString(from: save(context)) + } +} diff --git a/runtime/swift/prompty-model/Tests/PromptyModelTests/Contracts/Events/StreamFailureTests.swift b/runtime/swift/prompty-model/Tests/PromptyModelTests/Contracts/Events/StreamFailureTests.swift new file mode 100644 index 000000000..68d8d6e62 --- /dev/null +++ b/runtime/swift/prompty-model/Tests/PromptyModelTests/Contracts/Events/StreamFailureTests.swift @@ -0,0 +1,38 @@ +// +// Code generated by Typra emitter; DO NOT EDIT. + +import Foundation +import XCTest +@testable import PromptyModel + +final class StreamFailureTests: XCTestCase { + func testJSONRoundTrip1() throws { + let json = """ +{ + "outcome": "indeterminate", + "message": "SSE stream error: connection reset" +} +""" + let instance = try StreamFailure.fromJSON(json) + XCTAssertEqual(instance.outcome, StreamFailureOutcome.indeterminate) + XCTAssertEqual(instance.message, "SSE stream error: connection reset") + let reloaded = try StreamFailure.fromJSON(try instance.toJSON()) + XCTAssertEqual(reloaded.outcome, StreamFailureOutcome.indeterminate) + XCTAssertEqual(reloaded.message, "SSE stream error: connection reset") + } + + func testYAMLRoundTrip1() throws { + let yaml = """ +outcome: indeterminate +message: "SSE stream error: connection reset" + +""" + let instance = try StreamFailure.fromYAML(yaml) + XCTAssertEqual(instance.outcome, StreamFailureOutcome.indeterminate) + XCTAssertEqual(instance.message, "SSE stream error: connection reset") + let reloaded = try StreamFailure.fromYAML(try instance.toYAML()) + XCTAssertEqual(reloaded.outcome, StreamFailureOutcome.indeterminate) + XCTAssertEqual(reloaded.message, "SSE stream error: connection reset") + } + +} diff --git a/runtime/typescript/packages/anthropic/src/processor.ts b/runtime/typescript/packages/anthropic/src/processor.ts index e87fcabae..bad6443cc 100644 --- a/runtime/typescript/packages/anthropic/src/processor.ts +++ b/runtime/typescript/packages/anthropic/src/processor.ts @@ -13,13 +13,17 @@ import type { Agent } from "@prompty/core"; import type { Processor } from "@prompty/core"; import type { ToolCall } from "@prompty/core"; +import { FailureChunk, StreamFailure } from "@prompty/core"; import { traceSpan } from "@prompty/core"; import { createStructuredResult } from "@prompty/core"; export class AnthropicProcessor implements Processor { async process(agent: Agent, response: unknown): Promise { return traceSpan("AnthropicProcessor", async (emit) => { - emit("signature", "prompty.anthropic.processor.AnthropicProcessor.invoke"); + emit( + "signature", + "prompty.anthropic.processor.AnthropicProcessor.invoke", + ); emit("inputs", { data: response }); const result = processResponse(agent, response); // Don't emit result for streaming — it's a generator, not a value @@ -59,9 +63,7 @@ export function processResponse(agent: Agent, response: unknown): unknown { /** Type guard for async iterables (PromptyStream or raw SDK stream). */ function isAsyncIterable(value: unknown): value is AsyncIterable { return ( - typeof value === "object" && - value !== null && - Symbol.asyncIterator in value + typeof value === "object" && value !== null && Symbol.asyncIterator in value ); } @@ -78,48 +80,96 @@ function isAsyncIterable(value: unknown): value is AsyncIterable { */ async function* streamGenerator( response: AsyncIterable, -): AsyncGenerator { +): AsyncGenerator { const toolCallAcc: Map< number, { id: string; name: string; arguments: string } > = new Map(); + const iterator = response[Symbol.asyncIterator](); + let iteratorClosed = false; + let iteratorExhausted = false; + + const close = async (): Promise => { + if (iteratorClosed) return; + iteratorClosed = true; + await closeIterator(iterator); + }; + + try { + while (true) { + let next: IteratorResult; + try { + next = await iterator.next(); + } catch (error) { + await close(); + yield new FailureChunk({ + failure: new StreamFailure({ + outcome: "indeterminate", + message: error instanceof Error ? error.message : String(error), + }), + }); + return; + } + if (next.done) { + iteratorExhausted = true; + break; + } - for await (const event of response) { - const e = event as Record; - const eventType = e.type as string | undefined; - - if (eventType === "content_block_delta") { - const delta = e.delta as Record | undefined; - if (!delta) continue; - - if (delta.type === "text_delta") { - yield delta.text as string; - } else if (delta.type === "input_json_delta") { - // Accumulate partial JSON for tool arguments - const idx = e.index as number; - const acc = toolCallAcc.get(idx); - if (acc) { - acc.arguments += (delta.partial_json ?? "") as string; + const event = next.value; + const e = event as Record; + const eventType = e.type as string | undefined; + + if (eventType === "content_block_delta") { + const delta = e.delta as Record | undefined; + if (!delta) continue; + + if (delta.type === "text_delta") { + yield delta.text as string; + } else if (delta.type === "input_json_delta") { + // Accumulate partial JSON for tool arguments + const idx = e.index as number; + const acc = toolCallAcc.get(idx); + if (acc) { + acc.arguments += (delta.partial_json ?? "") as string; + } + } + } else if (eventType === "content_block_start") { + const block = e.content_block as Record | undefined; + if (block?.type === "tool_use") { + const idx = e.index as number; + toolCallAcc.set(idx, { + id: (block.id ?? "") as string, + name: (block.name ?? "") as string, + arguments: "", + }); } - } - } else if (eventType === "content_block_start") { - const block = e.content_block as Record | undefined; - if (block?.type === "tool_use") { - const idx = e.index as number; - toolCallAcc.set(idx, { - id: (block.id ?? "") as string, - name: (block.name ?? "") as string, - arguments: "", - }); } } + + // Yield accumulated tool calls at the end of the stream + const sortedIndices = [...toolCallAcc.keys()].sort((a, b) => a - b); + for (const idx of sortedIndices) { + const tc = toolCallAcc.get(idx)!; + yield { id: tc.id, name: tc.name, arguments: tc.arguments } as ToolCall; + } + } finally { + if (!iteratorExhausted) { + await close(); + } } +} - // Yield accumulated tool calls at the end of the stream - const sortedIndices = [...toolCallAcc.keys()].sort((a, b) => a - b); - for (const idx of sortedIndices) { - const tc = toolCallAcc.get(idx)!; - yield { id: tc.id, name: tc.name, arguments: tc.arguments } as ToolCall; +async function closeIterator(iterator: AsyncIterator): Promise { + if (!iterator.return) return; + try { + await iterator.return(); + } catch (error) { + if (typeof globalThis.console?.debug === "function") { + globalThis.console.debug( + "Failed to close Anthropic response stream:", + error, + ); + } } } @@ -180,7 +230,10 @@ function processMessages( // Structured output — JSON parse when outputs schema exists if (agent.outputs && agent.outputs.length > 0) { try { - return createStructuredResult(JSON.parse(text) as Record, text); + return createStructuredResult( + JSON.parse(text) as Record, + text, + ); } catch { return text; } diff --git a/runtime/typescript/packages/anthropic/tests/stream-failures.test.ts b/runtime/typescript/packages/anthropic/tests/stream-failures.test.ts new file mode 100644 index 000000000..c3769e03c --- /dev/null +++ b/runtime/typescript/packages/anthropic/tests/stream-failures.test.ts @@ -0,0 +1,71 @@ +import { FailureChunk, Agent } from "@prompty/core"; +import { describe, expect, it } from "vitest"; + +import { processResponse } from "../src/processor.js"; + +describe("Anthropic classified stream failures", () => { + it("closes the provider stream when the consumer stops early", async () => { + let closeCount = 0; + const response: AsyncIterable = { + [Symbol.asyncIterator](): AsyncIterator { + return { + async next(): Promise> { + return { + done: false, + value: { + type: "content_block_delta", + delta: { type: "text_delta", text: "partial" }, + }, + }; + }, + async return(): Promise> { + closeCount += 1; + return { done: true, value: undefined }; + }, + }; + }, + }; + const agent = new Agent({ name: "stream-cancel", model: "claude-test" }); + + for await (const item of processResponse( + agent, + response, + ) as AsyncIterable) { + expect(item).toBe("partial"); + break; + } + + expect(closeCount).toBe(1); + }); + + it("closes the provider stream after a transport failure", async () => { + let closed = false; + const response: AsyncIterable = { + [Symbol.asyncIterator](): AsyncIterator { + return { + async next(): Promise> { + throw new Error("SSE stream error: connection reset"); + }, + async return(): Promise> { + closed = true; + return { done: true, value: undefined }; + }, + }; + }, + }; + const agent = new Agent({ name: "stream-failure", model: "claude-test" }); + const processed = processResponse(agent, response); + const chunks: unknown[] = []; + + for await (const chunk of processed as AsyncIterable) { + chunks.push(chunk); + } + + expect(closed).toBe(true); + expect(chunks).toHaveLength(1); + expect(chunks[0]).toBeInstanceOf(FailureChunk); + const failure = chunks[0] as FailureChunk; + expect(failure.failure.outcome).toBe("indeterminate"); + expect(failure.failure.message).toBe("SSE stream error: connection reset"); + }); +}); diff --git a/runtime/typescript/packages/core/src/core/index.ts b/runtime/typescript/packages/core/src/core/index.ts index 8f12035f1..725d300f5 100644 --- a/runtime/typescript/packages/core/src/core/index.ts +++ b/runtime/typescript/packages/core/src/core/index.ts @@ -14,6 +14,7 @@ export { invoke, resolveBindings, ExecuteError, + StreamFailureError, type TurnOptions, type InvokeOptions, } from "./pipeline.js"; diff --git a/runtime/typescript/packages/core/src/core/pipeline.ts b/runtime/typescript/packages/core/src/core/pipeline.ts index 147f75937..108c1fba5 100644 --- a/runtime/typescript/packages/core/src/core/pipeline.ts +++ b/runtime/typescript/packages/core/src/core/pipeline.ts @@ -32,6 +32,8 @@ */ import { Agent } from "../model/agent.js"; +import { FailureChunk } from "../model/contracts/events/stream-chunk.js"; +import { StreamFailure } from "../model/contracts/events/stream-failure.js"; import { type ToolCall, Message, @@ -92,6 +94,24 @@ export class ExecuteError extends Error { } } +/** Terminal classified failure from a processed model stream. */ +export class StreamFailureError extends Error { + public readonly failure: StreamFailure; + public readonly partialContent: string; + + constructor(failure: StreamFailure, partialContent: string) { + super(failure.message); + this.name = "StreamFailureError"; + this.failure = failure; + this.partialContent = partialContent; + } + + /** Whether the provider outcome must be reconciled before another invocation. */ + get requiresReconciliation(): boolean { + return this.failure.outcome === "indeterminate"; + } +} + /** Replace raw nonce strings with readable `{{thread:name}}` in trace output. */ function sanitizeNonces(value: unknown): unknown { const nonces = getLastNonces(); @@ -766,9 +786,24 @@ export async function turn( } let processed: unknown; try { - processed = await process(agent, response); + if (isAsyncIterable(response)) { + const streamResult = await consumeStream(agent, response, onEvent); + processed = streamResult.content; + } else { + processed = await process(agent, response); + } } catch (err) { - emitFailedTurnEnd(onEvent, err, 0, response); + if (err instanceof StreamFailureError) { + emitEvent(onEvent, "error", { + message: err.message, + outcome: err.failure.outcome, + requiresReconciliation: err.requiresReconciliation, + partialContent: err.partialContent, + }); + emitFailedTurnEnd(onEvent, err, 0, err.partialContent); + } else { + emitFailedTurnEnd(onEvent, err, 0, response); + } throw err; } @@ -893,7 +928,17 @@ export async function turn( try { streamResult = await consumeStream(agent, response, onEvent); } catch (err) { - emitFailedTurnEnd(onEvent, err, iteration, response); + if (err instanceof StreamFailureError) { + emitEvent(onEvent, "error", { + message: err.message, + outcome: err.failure.outcome, + requiresReconciliation: err.requiresReconciliation, + partialContent: err.partialContent, + }); + emitFailedTurnEnd(onEvent, err, iteration, err.partialContent); + } else { + emitFailedTurnEnd(onEvent, err, iteration, response); + } throw err; } const { toolCalls, content } = streamResult; @@ -1097,7 +1142,10 @@ async function consumeStream( if (isAsyncIterable(processed)) { for await (const item of processed) { - if (isToolCallLike(item)) { + const failure = streamFailureFrom(item); + if (failure !== undefined) { + throw new StreamFailureError(failure, textParts.join("")); + } else if (isToolCallLike(item)) { toolCalls.push(item); } else if (typeof item === "string") { textParts.push(item); @@ -1112,6 +1160,21 @@ async function consumeStream( return { toolCalls, content: textParts.join("") }; } +function streamFailureFrom(item: unknown): StreamFailure | undefined { + if (item instanceof FailureChunk) { + return item.failure; + } + if (typeof item !== "object" || item === null) { + return undefined; + } + + const chunk = item as Record; + if (chunk.kind !== "failure" || typeof chunk.failure !== "object" || chunk.failure === null) { + return undefined; + } + return StreamFailure.load(chunk.failure as Record); +} + // --------------------------------------------------------------------------- // Thread marker helpers diff --git a/runtime/typescript/packages/core/src/index.ts b/runtime/typescript/packages/core/src/index.ts index 0f2591d79..3b8acf999 100644 --- a/runtime/typescript/packages/core/src/index.ts +++ b/runtime/typescript/packages/core/src/index.ts @@ -69,6 +69,7 @@ export { turn, invoke, resolveBindings, + StreamFailureError, type TurnOptions, type InvokeOptions, @@ -197,6 +198,10 @@ export { PermissionDecision, HostToolRequest, HostToolResult, + StreamChunk, + ErrorChunk, + FailureChunk, + StreamFailure, type EventJournalWriter, type EventSink, type PermissionResolver, diff --git a/runtime/typescript/packages/core/src/model/contracts/events/index.ts b/runtime/typescript/packages/core/src/model/contracts/events/index.ts index ff991d35a..297037c9a 100644 --- a/runtime/typescript/packages/core/src/model/contracts/events/index.ts +++ b/runtime/typescript/packages/core/src/model/contracts/events/index.ts @@ -45,6 +45,7 @@ export { SessionFileRef } from "./session-file-ref"; export { SessionRef } from "./session-ref"; export { SessionSummary } from "./session-summary"; export { SessionTrace } from "./session-trace"; +export { StreamFailure } from "./stream-failure"; export { StreamChunk, TextChunk, @@ -52,4 +53,5 @@ export { ToolChunk, UsageChunk, ErrorChunk, + FailureChunk, } from "./stream-chunk"; diff --git a/runtime/typescript/packages/core/src/model/contracts/events/stream-chunk.ts b/runtime/typescript/packages/core/src/model/contracts/events/stream-chunk.ts index 4d15b6f44..24df00987 100644 --- a/runtime/typescript/packages/core/src/model/contracts/events/stream-chunk.ts +++ b/runtime/typescript/packages/core/src/model/contracts/events/stream-chunk.ts @@ -4,6 +4,7 @@ import { LoadContext, SaveContext } from "../../context"; import { InvocationUsage } from "../../contracts/models/invocation-usage"; +import { StreamFailure } from "./stream-failure"; import { ToolCall } from "../../contracts/conversation/tool-call"; export abstract class StreamChunk { @@ -66,6 +67,8 @@ export abstract class StreamChunk { return UsageChunk.load(data, context); case "error": return ErrorChunk.load(data, context); + case "failure": + return FailureChunk.load(data, context); default: throw new Error( `Unknown StreamChunk discriminator field 'kind' value: ${discriminator}`, @@ -548,3 +551,95 @@ export class ErrorChunk extends StreamChunk { //#endregion } + +export class FailureChunk extends StreamChunk { + static readonly shorthandProperty: string | undefined = undefined; + + kind: string = "failure"; + failure!: StreamFailure; + + constructor(init?: Partial) { + super(init); + this.kind = init?.kind ?? "failure"; + if (init?.failure !== undefined) { + this.failure = init.failure; + } + } + + //#region Load Methods + + static load( + data: Record, + context?: LoadContext, + ): FailureChunk { + context ??= new LoadContext(); + if (context) { + data = context.processInput(data) as Record; + } + + if (data["failure"] === undefined || data["failure"] === null) { + throw new Error(`${context.at("failure").path}: missing required field`); + } + const instance = new FailureChunk(); + + if (data["kind"] !== undefined && data["kind"] !== null) { + instance.kind = String(data["kind"]); + } + if (data["failure"] !== undefined && data["failure"] !== null) { + instance.failure = StreamFailure.load( + data["failure"] as Record, + context.at("failure"), + ); + } + + if (context) { + return context.processOutput(instance) as FailureChunk; + } + return instance; + } + + //#endregion + + //#region Save Methods + + save(context?: SaveContext): Record { + let obj: this = this; + if (context) { + obj = context.processObject(obj) as this; + } + + // Start with parent class properties + const result = super.save(context); + + if (obj.kind !== undefined && obj.kind !== null) { + result["kind"] = obj.kind; + } + if (obj.failure !== undefined && obj.failure !== null) { + result["failure"] = obj.failure.save(context); + } + return result; + } + + toYaml(context?: SaveContext): string { + context = context ?? new SaveContext(); + return context.toYaml(this.save(context)); + } + + toJson(context?: SaveContext, indent: number = 2): string { + context = context ?? new SaveContext(); + return context.toJson(this.save(context), indent); + } + + static fromJson(json: string, context?: LoadContext): FailureChunk { + const data = JSON.parse(json); + return FailureChunk.load(data as Record, context); + } + + static fromYaml(yaml: string, context?: LoadContext): FailureChunk { + const { parse } = require("yaml"); + const data = parse(yaml); + return FailureChunk.load(data as Record, context); + } + + //#endregion +} diff --git a/runtime/typescript/packages/core/src/model/contracts/events/stream-failure.ts b/runtime/typescript/packages/core/src/model/contracts/events/stream-failure.ts new file mode 100644 index 000000000..b51832c2e --- /dev/null +++ b/runtime/typescript/packages/core/src/model/contracts/events/stream-failure.ts @@ -0,0 +1,93 @@ +// +// Copyright (c) Microsoft. All rights reserved. +// WARNING: This is an auto-generated file. DO NOT EDIT THIS FILE DIRECTLY. + +import { LoadContext, SaveContext } from "../../context"; + +export type StreamFailureOutcome = "determinate" | "indeterminate"; + +export class StreamFailure { + static readonly shorthandProperty: string | undefined = undefined; + + outcome: StreamFailureOutcome = "determinate"; + message: string = ""; + + constructor(init?: Partial) { + this.outcome = init?.outcome ?? "determinate"; + this.message = init?.message ?? ""; + } + + //#region Load Methods + + static load( + data: Record, + context?: LoadContext, + ): StreamFailure { + context ??= new LoadContext(); + if (context) { + data = context.processInput(data) as Record; + } + + const instance = new StreamFailure(); + + if (data["outcome"] !== undefined && data["outcome"] !== null) { + instance.outcome = String(data["outcome"]) as StreamFailureOutcome; + } + if (data["message"] !== undefined && data["message"] !== null) { + instance.message = String(data["message"]); + } + + if (context) { + return context.processOutput(instance) as StreamFailure; + } + return instance; + } + + //#endregion + + //#region Save Methods + + save(context?: SaveContext): Record { + let obj: this = this; + if (context) { + obj = context.processObject(obj) as this; + } + + const result: Record = {}; + + if (obj.outcome !== undefined && obj.outcome !== null) { + result["outcome"] = obj.outcome; + } + if (obj.message !== undefined && obj.message !== null) { + result["message"] = obj.message; + } + + if (context) { + return context.processDict(result); + } + return result; + } + + toYaml(context?: SaveContext): string { + context = context ?? new SaveContext(); + return context.toYaml(this.save(context)); + } + + toJson(context?: SaveContext, indent: number = 2): string { + context = context ?? new SaveContext(); + return context.toJson(this.save(context), indent); + } + + static fromJson(json: string, context?: LoadContext): StreamFailure { + const data = JSON.parse(json); + return StreamFailure.load(data as Record, context); + } + + static fromYaml(yaml: string, context?: LoadContext): StreamFailure { + const { parse } = require("yaml"); + const data = parse(yaml); + return StreamFailure.load(data as Record, context); + } + + //#endregion +} diff --git a/runtime/typescript/packages/core/src/model/index.ts b/runtime/typescript/packages/core/src/model/index.ts index aa623dce4..6434969f5 100644 --- a/runtime/typescript/packages/core/src/model/index.ts +++ b/runtime/typescript/packages/core/src/model/index.ts @@ -102,6 +102,7 @@ export { SessionFileRef } from "./contracts/events/session-file-ref"; export { SessionRef } from "./contracts/events/session-ref"; export { SessionSummary } from "./contracts/events/session-summary"; export { SessionTrace } from "./contracts/events/session-trace"; +export { StreamFailure } from "./contracts/events/stream-failure"; export { StreamChunk, TextChunk, @@ -109,6 +110,7 @@ export { ToolChunk, UsageChunk, ErrorChunk, + FailureChunk, } from "./contracts/events/stream-chunk"; export { GuardrailResult } from "./contracts/guardrails/guardrail-result"; diff --git a/runtime/typescript/packages/core/tests/model/contracts/events/failure-chunk.test.ts b/runtime/typescript/packages/core/tests/model/contracts/events/failure-chunk.test.ts new file mode 100644 index 000000000..c6ac814aa --- /dev/null +++ b/runtime/typescript/packages/core/tests/model/contracts/events/failure-chunk.test.ts @@ -0,0 +1,68 @@ +// +// Copyright (c) Microsoft. All rights reserved. +// WARNING: This is an auto-generated file. DO NOT EDIT THIS FILE DIRECTLY. + +import { FailureChunk } from "../../../../src/model/index"; + +describe("FailureChunk", () => { + describe("construction", () => { + it("should create a new instance with defaults", () => { + const instance = new FailureChunk(); + expect(instance).toBeDefined(); + }); + + it("should create a new instance with partial initialization", () => { + const instance = new FailureChunk({}); + expect(instance).toBeDefined(); + }); + }); + + describe("JSON serialization", () => { + it("should load from JSON - example 1", () => { + const json = `{\n "failure": {\n "outcome": "indeterminate",\n "message": "SSE stream error: connection reset"\n }\n}`; + const instance = FailureChunk.fromJson(json); + expect(instance).toBeDefined(); + }); + + it("should round-trip JSON - example 1", () => { + const json = `{\n "failure": {\n "outcome": "indeterminate",\n "message": "SSE stream error: connection reset"\n }\n}`; + const instance = FailureChunk.fromJson(json); + const output = instance.toJson(); + const reloaded = FailureChunk.fromJson(output); + expect(reloaded).toBeDefined(); + }); + }); + + describe("YAML serialization", () => { + it("should load from YAML - example 1", () => { + const yaml = `failure:\n outcome: indeterminate\n message: "SSE stream error: connection reset"\n`; + const instance = FailureChunk.fromYaml(yaml); + expect(instance).toBeDefined(); + }); + + it("should round-trip YAML - example 1", () => { + const yaml = `failure:\n outcome: indeterminate\n message: "SSE stream error: connection reset"\n`; + const instance = FailureChunk.fromYaml(yaml); + const output = instance.toYaml(); + const reloaded = FailureChunk.fromYaml(output); + expect(reloaded).toBeDefined(); + }); + }); + + describe("load and save", () => { + it("should load from dictionary", () => { + const data = JSON.parse( + `{\n "failure": {\n "outcome": "indeterminate",\n "message": "SSE stream error: connection reset"\n }\n}`, + ) as Record; + const instance = FailureChunk.load(data); + expect(instance).toBeDefined(); + }); + + it("should save to dictionary", () => { + const instance = new FailureChunk(); + const data = instance.save(); + expect(data).toBeDefined(); + expect(typeof data).toBe("object"); + }); + }); +}); diff --git a/runtime/typescript/packages/core/tests/model/contracts/events/stream-failure.test.ts b/runtime/typescript/packages/core/tests/model/contracts/events/stream-failure.test.ts new file mode 100644 index 000000000..48c8897b7 --- /dev/null +++ b/runtime/typescript/packages/core/tests/model/contracts/events/stream-failure.test.ts @@ -0,0 +1,74 @@ +// +// Copyright (c) Microsoft. All rights reserved. +// WARNING: This is an auto-generated file. DO NOT EDIT THIS FILE DIRECTLY. + +import { StreamFailure } from "../../../../src/model/index"; + +describe("StreamFailure", () => { + describe("construction", () => { + it("should create a new instance with defaults", () => { + const instance = new StreamFailure(); + expect(instance).toBeDefined(); + }); + + it("should create a new instance with partial initialization", () => { + const instance = new StreamFailure({}); + expect(instance).toBeDefined(); + }); + }); + + describe("JSON serialization", () => { + it("should load from JSON - example 1", () => { + const json = `{\n "outcome": "indeterminate",\n "message": "SSE stream error: connection reset"\n}`; + const instance = StreamFailure.fromJson(json); + expect(instance).toBeDefined(); + expect(instance.outcome).toEqual("indeterminate"); + expect(instance.message).toEqual("SSE stream error: connection reset"); + }); + + it("should round-trip JSON - example 1", () => { + const json = `{\n "outcome": "indeterminate",\n "message": "SSE stream error: connection reset"\n}`; + const instance = StreamFailure.fromJson(json); + const output = instance.toJson(); + const reloaded = StreamFailure.fromJson(output); + expect(reloaded.outcome).toEqual(instance.outcome); + expect(reloaded.message).toEqual(instance.message); + }); + }); + + describe("YAML serialization", () => { + it("should load from YAML - example 1", () => { + const yaml = `outcome: indeterminate\nmessage: "SSE stream error: connection reset"\n`; + const instance = StreamFailure.fromYaml(yaml); + expect(instance).toBeDefined(); + expect(instance.outcome).toEqual("indeterminate"); + expect(instance.message).toEqual("SSE stream error: connection reset"); + }); + + it("should round-trip YAML - example 1", () => { + const yaml = `outcome: indeterminate\nmessage: "SSE stream error: connection reset"\n`; + const instance = StreamFailure.fromYaml(yaml); + const output = instance.toYaml(); + const reloaded = StreamFailure.fromYaml(output); + expect(reloaded.outcome).toEqual(instance.outcome); + expect(reloaded.message).toEqual(instance.message); + }); + }); + + describe("load and save", () => { + it("should load from dictionary", () => { + const data = JSON.parse( + `{\n "outcome": "indeterminate",\n "message": "SSE stream error: connection reset"\n}`, + ) as Record; + const instance = StreamFailure.load(data); + expect(instance).toBeDefined(); + }); + + it("should save to dictionary", () => { + const instance = new StreamFailure(); + const data = instance.save(); + expect(data).toBeDefined(); + expect(typeof data).toBe("object"); + }); + }); +}); diff --git a/runtime/typescript/packages/core/tests/stream-failures.test.ts b/runtime/typescript/packages/core/tests/stream-failures.test.ts new file mode 100644 index 000000000..f6b3dab0f --- /dev/null +++ b/runtime/typescript/packages/core/tests/stream-failures.test.ts @@ -0,0 +1,88 @@ +import { readFileSync } from "node:fs"; +import { resolve } from "node:path"; + +import { OpenAIProcessor } from "@prompty/openai"; +import { beforeEach, describe, expect, it } from "vitest"; + +import type { EventCallback } from "../src/core/agent-events.js"; +import type { Executor } from "../src/core/interfaces.js"; +import { StreamFailureError, turn } from "../src/core/pipeline.js"; +import { + clearCache, + registerExecutor, + registerParser, + registerProcessor, + registerRenderer, +} from "../src/core/registry.js"; +import { Agent } from "../src/model/agent.js"; +import { PromptyChatParser } from "../src/parsers/prompty.js"; +import { NunjucksRenderer } from "../src/renderers/nunjucks.js"; + +interface StreamFailureVector { + name: string; + input: { + events: Array<{ kind: "provider"; value: Record } | { kind: "transportError"; message: string }>; + }; + expected: { partialText: string; requiresReconciliation: boolean; completionCommitted: boolean }; +} + +const PROVIDER = "stream-failure-vector"; + +function loadVectors(): StreamFailureVector[] { + const path = resolve(import.meta.dirname, "../../../../../spec/vectors/process/stream_failure_vectors.json"); + return JSON.parse(readFileSync(path, "utf8")) as StreamFailureVector[]; +} + +class VectorExecutor implements Executor { + constructor(private readonly vector: StreamFailureVector) {} + + async execute(): Promise> { + const events = this.vector.input.events; + return { + async *[Symbol.asyncIterator](): AsyncIterator { + for (const event of events) { + if (event.kind === "transportError") { + throw new Error(event.message); + } + yield event.value; + } + }, + }; + } +} + +function makeAgent(): Agent { + return Agent.load({ + name: "stream-failure-vector", + model: { id: "gpt-test", provider: PROVIDER }, + instructions: "user:\nHello", + }); +} + +describe("turn classified stream failures", () => { + beforeEach(() => { + clearCache(); + registerRenderer("nunjucks", new NunjucksRenderer()); + registerParser("prompty", new PromptyChatParser()); + }); + + for (const vector of loadVectors()) { + it(vector.name, async () => { + registerExecutor(PROVIDER, new VectorExecutor(vector)); + registerProcessor(PROVIDER, new OpenAIProcessor()); + const events: Array<{ type: string; data: Record }> = []; + const onEvent: EventCallback = (type, data) => events.push({ type, data }); + + const error = await turn(makeAgent(), {}, { onEvent }).catch((caught: unknown) => caught); + + expect(error).toBeInstanceOf(StreamFailureError); + const failure = error as StreamFailureError; + expect(failure.partialContent).toBe(vector.expected.partialText); + expect(failure.requiresReconciliation).toBe(vector.expected.requiresReconciliation); + expect(events.some(event => event.type === "done")).toBe(vector.expected.completionCommitted); + expect(events.some(event => event.type === "turn_end" && event.data.status === "success")).toBe( + vector.expected.completionCommitted, + ); + }); + } +}); diff --git a/runtime/typescript/packages/openai/src/processor.ts b/runtime/typescript/packages/openai/src/processor.ts index 18512c0ad..ef58ab95f 100644 --- a/runtime/typescript/packages/openai/src/processor.ts +++ b/runtime/typescript/packages/openai/src/processor.ts @@ -9,6 +9,7 @@ import type { Agent } from "@prompty/core"; import type { Processor } from "@prompty/core"; import type { ToolCall } from "@prompty/core"; +import { FailureChunk, StreamFailure } from "@prompty/core"; import { traceSpan } from "@prompty/core"; import { createStructuredResult } from "@prompty/core"; @@ -73,9 +74,7 @@ export function processResponse(agent: Agent, response: unknown): unknown { /** Type guard for async iterables (PromptyStream or raw SDK stream). */ function isAsyncIterable(value: unknown): value is AsyncIterable { return ( - typeof value === "object" && - value !== null && - Symbol.asyncIterator in value + typeof value === "object" && value !== null && Symbol.asyncIterator in value ); } @@ -86,57 +85,122 @@ function isAsyncIterable(value: unknown): value is AsyncIterable { * - `delta.content` — yields content strings * - `delta.tool_calls` — accumulates partial tool call chunks, * yields ToolCall objects when the stream ends - * - `delta.refusal` — throws Error with the refusal message + * - `delta.refusal` — yields a determinate FailureChunk * * Matches the Python `_stream_generator` / `_async_stream_generator`. */ async function* streamGenerator( response: AsyncIterable, -): AsyncGenerator { - const toolCallAcc: Map = new Map(); +): AsyncGenerator { + const toolCallAcc: Map< + number, + { id: string; name: string; arguments: string } + > = new Map(); + const iterator = response[Symbol.asyncIterator](); + let iteratorClosed = false; + let iteratorExhausted = false; + + const close = async (): Promise => { + if (iteratorClosed) return; + iteratorClosed = true; + await closeIterator(iterator); + }; + + try { + while (true) { + let next: IteratorResult; + try { + next = await iterator.next(); + } catch (error) { + await close(); + yield failureChunk("indeterminate", errorMessage(error)); + return; + } + if (next.done) { + iteratorExhausted = true; + break; + } - for await (const chunk of response) { - const c = chunk as Record; - const choices = c.choices as Record[] | undefined; - if (!choices || choices.length === 0) continue; + const chunk = next.value; + const c = chunk as Record; + const choices = c.choices as Record[] | undefined; + if (!choices || choices.length === 0) continue; - const delta = (choices[0] as Record).delta as Record | undefined; - if (!delta) continue; + const delta = (choices[0] as Record).delta as + | Record + | undefined; + if (!delta) continue; - // Content - if (delta.content != null) { - yield delta.content as string; - } + // Content + if (delta.content != null) { + yield delta.content as string; + } - // Tool call deltas — accumulate index-keyed partial chunks - const tcDeltas = delta.tool_calls as Record[] | undefined; - if (tcDeltas) { - for (const tcDelta of tcDeltas) { - const idx = tcDelta.index as number; - if (!toolCallAcc.has(idx)) { - toolCallAcc.set(idx, { id: "", name: "", arguments: "" }); - } - const acc = toolCallAcc.get(idx)!; - if (tcDelta.id) acc.id = tcDelta.id as string; - const fn = tcDelta.function as Record | undefined; - if (fn) { - if (fn.name) acc.name = fn.name as string; - if (fn.arguments) acc.arguments += fn.arguments as string; + // Tool call deltas — accumulate index-keyed partial chunks + const tcDeltas = delta.tool_calls as + | Record[] + | undefined; + if (tcDeltas) { + for (const tcDelta of tcDeltas) { + const idx = tcDelta.index as number; + if (!toolCallAcc.has(idx)) { + toolCallAcc.set(idx, { id: "", name: "", arguments: "" }); + } + const acc = toolCallAcc.get(idx)!; + if (tcDelta.id) acc.id = tcDelta.id as string; + const fn = tcDelta.function as Record | undefined; + if (fn) { + if (fn.name) acc.name = fn.name as string; + if (fn.arguments) acc.arguments += fn.arguments as string; + } } } + + // Refusal + if (delta.refusal != null) { + await close(); + yield failureChunk("determinate", `Model refused: ${delta.refusal}`); + return; + } } - // Refusal - if (delta.refusal != null) { - throw new Error(`Model refused: ${delta.refusal}`); + // Yield accumulated tool calls at the end of the stream + const sortedIndices = [...toolCallAcc.keys()].sort((a, b) => a - b); + for (const idx of sortedIndices) { + const tc = toolCallAcc.get(idx)!; + yield { id: tc.id, name: tc.name, arguments: tc.arguments } as ToolCall; + } + } finally { + if (!iteratorExhausted) { + await close(); } } +} + +function failureChunk( + outcome: "determinate" | "indeterminate", + message: string, +): FailureChunk { + return new FailureChunk({ + failure: new StreamFailure({ outcome, message }), + }); +} + +function errorMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} - // Yield accumulated tool calls at the end of the stream - const sortedIndices = [...toolCallAcc.keys()].sort((a, b) => a - b); - for (const idx of sortedIndices) { - const tc = toolCallAcc.get(idx)!; - yield { id: tc.id, name: tc.name, arguments: tc.arguments } as ToolCall; +async function closeIterator(iterator: AsyncIterator): Promise { + if (!iterator.return) return; + try { + await iterator.return(); + } catch (error) { + if (typeof globalThis.console?.debug === "function") { + globalThis.console.debug( + "Failed to close OpenAI response stream:", + error, + ); + } } } @@ -180,7 +244,10 @@ function processResponsesApi( // Structured output — JSON parse when outputs schema exists if (agent.outputs && agent.outputs.length > 0) { try { - return createStructuredResult(JSON.parse(outputText) as Record, outputText); + return createStructuredResult( + JSON.parse(outputText) as Record, + outputText, + ); } catch { return outputText; } @@ -207,7 +274,10 @@ function processResponsesApi( const text = texts.join(""); if (agent.outputs && agent.outputs.length > 0) { try { - return createStructuredResult(JSON.parse(text) as Record, text); + return createStructuredResult( + JSON.parse(text) as Record, + text, + ); } catch { return text; } @@ -259,7 +329,10 @@ function processChatCompletion( // Structured output — JSON parse when outputs schema exists if (agent.outputs && agent.outputs.length > 0) { try { - return createStructuredResult(JSON.parse(content) as Record, content); + return createStructuredResult( + JSON.parse(content) as Record, + content, + ); } catch { return content; } diff --git a/runtime/typescript/packages/openai/tests/stream-failure-vectors.test.ts b/runtime/typescript/packages/openai/tests/stream-failure-vectors.test.ts new file mode 100644 index 000000000..9874796b6 --- /dev/null +++ b/runtime/typescript/packages/openai/tests/stream-failure-vectors.test.ts @@ -0,0 +1,141 @@ +import { readFileSync } from "node:fs"; +import { resolve } from "node:path"; + +import { FailureChunk, Agent, StreamChunk } from "@prompty/core"; +import { describe, expect, it } from "vitest"; + +import { processResponse } from "../src/processor.js"; + +interface StreamFailureVector { + name: string; + input: { + provider: string; + events: Array< + | { kind: "provider"; value: Record } + | { kind: "transportError"; message: string } + >; + }; + expected: { chunks: unknown[] }; +} + +function loadVectors(): StreamFailureVector[] { + const path = resolve( + import.meta.dirname, + "../../../../../spec/vectors/process/stream_failure_vectors.json", + ); + return JSON.parse(readFileSync(path, "utf8")) as StreamFailureVector[]; +} + +function responseFromVector( + vector: StreamFailureVector, +): AsyncIterable { + return { + async *[Symbol.asyncIterator](): AsyncIterator { + for (const event of vector.input.events) { + if (event.kind === "transportError") { + throw new Error(event.message); + } + yield event.value; + } + }, + }; +} + +function closableResponseFromVector( + vector: StreamFailureVector, + onClose: () => void, +): AsyncIterable { + const events = vector.input.events; + let index = 0; + return { + [Symbol.asyncIterator](): AsyncIterator { + return { + async next(): Promise> { + const event = events[index++]; + if (event === undefined) return { done: true, value: undefined }; + if (event.kind === "transportError") { + throw new Error(event.message); + } + return { done: false, value: event.value }; + }, + async return(): Promise> { + onClose(); + return { done: true, value: undefined }; + }, + }; + }, + }; +} + +describe("OpenAI classified stream failure vectors", () => { + it("closes the provider stream when the consumer stops early", async () => { + let closeCount = 0; + const response: AsyncIterable = { + [Symbol.asyncIterator](): AsyncIterator { + return { + async next(): Promise> { + return { + done: false, + value: { choices: [{ delta: { content: "partial" } }] }, + }; + }, + async return(): Promise> { + closeCount += 1; + return { done: true, value: undefined }; + }, + }; + }, + }; + const agent = new Agent({ name: "stream-cancel", model: "gpt-test" }); + + for await (const item of processResponse( + agent, + response, + ) as AsyncIterable) { + expect(item).toBe("partial"); + break; + } + + expect(closeCount).toBe(1); + }); + + for (const vector of loadVectors()) { + it(vector.name, async () => { + expect(vector.input.provider).toBe("openai"); + const agent = new Agent({ name: "stream-vector", model: "gpt-test" }); + const processed = processResponse(agent, responseFromVector(vector)); + const actual: unknown[] = []; + + for await (const item of processed as AsyncIterable) { + if (item instanceof FailureChunk) { + const saved = item.save(); + const loaded = StreamChunk.load(saved); + expect(loaded).toBeInstanceOf(FailureChunk); + expect(loaded.save()).toEqual(saved); + actual.push(saved); + } else { + actual.push({ kind: "text", value: item }); + } + } + + expect(actual).toEqual(vector.expected.chunks); + }); + + it(`${vector.name} closes the provider stream`, async () => { + let closed = false; + const agent = new Agent({ name: "stream-vector", model: "gpt-test" }); + const processed = processResponse( + agent, + closableResponseFromVector(vector, () => { + closed = true; + }), + ); + + for await (const _ of processed as AsyncIterable) { + // Consume the terminal failure chunk. + } + + expect(closed).toBe(true); + }); + } +}); diff --git a/schema/model/conformance/samples/events/stream-chunks.tsp b/schema/model/conformance/samples/events/stream-chunks.tsp index 92ab789b6..7c896f549 100644 --- a/schema/model/conformance/samples/events/stream-chunks.tsp +++ b/schema/model/conformance/samples/events/stream-chunks.tsp @@ -17,3 +17,16 @@ const ToolChunkToolCallSample = #{ }; const ErrorChunkMessageSample = #{ message: "Rate limit exceeded" }; + +const StreamFailureOutcomeSample = #{ outcome: "indeterminate" }; + +const StreamFailureMessageSample = #{ + message: "SSE stream error: connection reset", +}; + +const FailureChunkFailureSample = #{ + failure: #{ + outcome: "indeterminate", + message: "SSE stream error: connection reset", + }, +}; diff --git a/schema/model/contracts/events/stream-chunks.tsp b/schema/model/contracts/events/stream-chunks.tsp index 65a3f9599..108396b2b 100644 --- a/schema/model/contracts/events/stream-chunks.tsp +++ b/schema/model/contracts/events/stream-chunks.tsp @@ -10,7 +10,31 @@ namespace Prompty.Contracts.Events; /** * The kind discriminator for stream chunks. */ -alias StreamChunkType = "text" | "thinking" | "tool" | "usage" | "error"; +alias StreamChunkType = + | "text" + | "thinking" + | "tool" + | "usage" + | "error" + | "failure"; + +/** + * Whether a stream failure has a known provider outcome. + */ +alias StreamFailureOutcome = "determinate" | "indeterminate"; + +/** + * A classified terminal failure from an LLM response stream. + */ +model StreamFailure { + @doc("Whether the provider outcome is known or requires reconciliation") + @sample(StreamFailureOutcomeSample) + outcome: StreamFailureOutcome; + + @doc("The human-readable failure message") + @sample(StreamFailureMessageSample) + message: string; +} /** * A chunk of data from a streaming LLM response. Stream chunks are @@ -81,3 +105,15 @@ model ErrorChunk extends StreamChunk { @sample(ErrorChunkMessageSample) message: string; } + +/** + * A classified failure chunk from the LLM response stream. + */ +model FailureChunk extends StreamChunk { + @doc("The kind identifier for classified failure chunks") + kind: "failure"; + + @doc("The classified stream failure") + @sample(FailureChunkFailureSample) + failure: StreamFailure; +} diff --git a/schema/tsp-output/.typra-generated/export-surfaces.json b/schema/tsp-output/.typra-generated/export-surfaces.json index e483d652e..6be910788 100644 --- a/schema/tsp-output/.typra-generated/export-surfaces.json +++ b/schema/tsp-output/.typra-generated/export-surfaces.json @@ -79,6 +79,7 @@ "EventJournalWriter", "EventSink", "Executor", + "FailureChunk", "FileNotFoundError", "FilePart", "FinalOutputPolicyRequest", @@ -158,6 +159,7 @@ "SessionWarningPayload", "StatusEventPayload", "StreamChunk", + "StreamFailure", "StreamOptions", "SubscriptionInfo", "Template", @@ -509,6 +511,13 @@ "source": "contracts/ErrorEventPayload.cs", "protocol": false }, + { + "name": "FailureChunk", + "kind": "value", + "group": "contracts", + "source": "contracts/StreamChunk.cs", + "protocol": false + }, { "name": "FileNotFoundError", "kind": "value", @@ -1013,6 +1022,13 @@ "source": "contracts/StreamChunk.cs", "protocol": false }, + { + "name": "StreamFailure", + "kind": "value", + "group": "contracts", + "source": "contracts/StreamFailure.cs", + "protocol": false + }, { "name": "StreamOptions", "kind": "value", @@ -1386,6 +1402,7 @@ "EnginePermissionDecision", "ErrorChunk", "ErrorEventPayload", + "FailureChunk", "FileNotFoundError", "FilePart", "FinalOutputPolicyRequest", @@ -1458,6 +1475,7 @@ "SessionWarningPayload", "StatusEventPayload", "StreamChunk", + "StreamFailure", "StreamOptions", "SubscriptionInfo", "Template", @@ -1579,6 +1597,7 @@ "SessionWarningPayload.cs", "StatusEventPayload.cs", "StreamChunk.cs", + "StreamFailure.cs", "StreamOptions.cs", "SubscriptionInfo.cs", "Template.cs", @@ -2274,6 +2293,7 @@ "contracts/SessionWarningPayload.cs", "contracts/StatusEventPayload.cs", "contracts/StreamChunk.cs", + "contracts/StreamFailure.cs", "contracts/StreamOptions.cs", "contracts/SubscriptionInfo.cs", "contracts/Template.cs", @@ -2369,6 +2389,7 @@ "EventJournalWriter", "EventSink", "Executor", + "FailureChunk", "FileNotFoundError", "FilePart", "FinalOutputPolicyRequest", @@ -2448,6 +2469,7 @@ "SessionWarningPayload", "StatusEventPayload", "StreamChunk", + "StreamFailure", "StreamOptions", "SubscriptionInfo", "Template", @@ -2799,6 +2821,13 @@ "source": "error_event_payload.go", "protocol": false }, + { + "name": "FailureChunk", + "kind": "value", + "group": "contracts", + "source": "stream_chunk.go", + "protocol": false + }, { "name": "FileNotFoundError", "kind": "value", @@ -3303,6 +3332,13 @@ "source": "stream_chunk.go", "protocol": false }, + { + "name": "StreamFailure", + "kind": "value", + "group": "contracts", + "source": "stream_failure.go", + "protocol": false + }, { "name": "StreamOptions", "kind": "value", @@ -3676,6 +3712,7 @@ "EnginePermissionDecision", "ErrorChunk", "ErrorEventPayload", + "FailureChunk", "FileNotFoundError", "FilePart", "FinalOutputPolicyRequest", @@ -3748,6 +3785,7 @@ "SessionWarningPayload", "StatusEventPayload", "StreamChunk", + "StreamFailure", "StreamOptions", "SubscriptionInfo", "Template", @@ -3869,6 +3907,7 @@ "session_warning_payload", "status_event_payload", "stream_chunk", + "stream_failure", "stream_options", "subscription_info", "template", @@ -4572,6 +4611,7 @@ "session_warning_payload.go", "status_event_payload.go", "stream_chunk.go", + "stream_failure.go", "stream_options.go", "subscription_info.go", "template.go", @@ -4658,6 +4698,7 @@ "EventJournalWriter", "EventSink", "Executor", + "FailureChunk", "FileNotFoundError", "FilePart", "FinalOutputPolicyRequest", @@ -4737,6 +4778,7 @@ "SessionWarningPayload", "StatusEventPayload", "StreamChunk", + "StreamFailure", "StreamOptions", "SubscriptionInfo", "Template", @@ -5088,6 +5130,13 @@ "source": "ErrorEventPayload", "protocol": false }, + { + "name": "FailureChunk", + "kind": "value", + "group": "contracts", + "source": "StreamChunk", + "protocol": false + }, { "name": "FileNotFoundError", "kind": "value", @@ -5592,6 +5641,13 @@ "source": "StreamChunk", "protocol": false }, + { + "name": "StreamFailure", + "kind": "value", + "group": "contracts", + "source": "StreamFailure", + "protocol": false + }, { "name": "StreamOptions", "kind": "value", @@ -5965,6 +6021,7 @@ "EnginePermissionDecision", "ErrorChunk", "ErrorEventPayload", + "FailureChunk", "FileNotFoundError", "FilePart", "FinalOutputPolicyRequest", @@ -6037,6 +6094,7 @@ "SessionWarningPayload", "StatusEventPayload", "StreamChunk", + "StreamFailure", "StreamOptions", "SubscriptionInfo", "Template", @@ -6158,6 +6216,7 @@ "session_warning_payload", "status_event_payload", "stream_chunk", + "stream_failure", "stream_options", "subscription_info", "template", @@ -6861,6 +6920,7 @@ "SessionWarningPayload", "StatusEventPayload", "StreamChunk", + "StreamFailure", "StreamOptions", "SubscriptionInfo", "Template", @@ -6948,6 +7008,7 @@ "EventJournalWriter", "EventSink", "Executor", + "FailureChunk", "FileNotFoundError", "FilePart", "FinalOutputPolicyRequest", @@ -7027,6 +7088,7 @@ "SessionWarningPayload", "StatusEventPayload", "StreamChunk", + "StreamFailure", "StreamOptions", "SubscriptionInfo", "Template", @@ -7378,6 +7440,13 @@ "source": ".contracts", "protocol": false }, + { + "name": "FailureChunk", + "kind": "value", + "group": "contracts", + "source": ".contracts", + "protocol": false + }, { "name": "FileNotFoundError", "kind": "value", @@ -7882,6 +7951,13 @@ "source": ".contracts", "protocol": false }, + { + "name": "StreamFailure", + "kind": "value", + "group": "contracts", + "source": ".contracts", + "protocol": false + }, { "name": "StreamOptions", "kind": "value", @@ -8255,6 +8331,7 @@ "EnginePermissionDecision", "ErrorChunk", "ErrorEventPayload", + "FailureChunk", "FileNotFoundError", "FilePart", "FinalOutputPolicyRequest", @@ -8327,6 +8404,7 @@ "SessionWarningPayload", "StatusEventPayload", "StreamChunk", + "StreamFailure", "StreamOptions", "SubscriptionInfo", "Template", @@ -8448,6 +8526,7 @@ "_SessionWarningPayload", "_StatusEventPayload", "_StreamChunk", + "_StreamFailure", "_StreamOptions", "_SubscriptionInfo", "_Template", @@ -9116,6 +9195,7 @@ "EventJournalWriter", "EventSink", "Executor", + "FailureChunk", "FileNotFoundError", "FilePart", "FinalOutputPolicyRequest", @@ -9195,6 +9275,7 @@ "SessionWarningPayload", "StatusEventPayload", "StreamChunk", + "StreamFailure", "StreamOptions", "SubscriptionInfo", "Template", @@ -9546,6 +9627,13 @@ "source": "contracts::error_event_payload", "protocol": false }, + { + "name": "FailureChunk", + "kind": "value", + "group": "contracts", + "source": "contracts::stream_chunk", + "protocol": false + }, { "name": "FileNotFoundError", "kind": "value", @@ -10050,6 +10138,13 @@ "source": "contracts::stream_chunk", "protocol": false }, + { + "name": "StreamFailure", + "kind": "value", + "group": "contracts", + "source": "contracts::stream_failure", + "protocol": false + }, { "name": "StreamOptions", "kind": "value", @@ -10423,6 +10518,7 @@ "EnginePermissionDecision", "ErrorChunk", "ErrorEventPayload", + "FailureChunk", "FileNotFoundError", "FilePart", "FinalOutputPolicyRequest", @@ -10495,6 +10591,7 @@ "SessionWarningPayload", "StatusEventPayload", "StreamChunk", + "StreamFailure", "StreamOptions", "SubscriptionInfo", "Template", @@ -10616,6 +10713,7 @@ "session_warning_payload", "status_event_payload", "stream_chunk", + "stream_failure", "stream_options", "subscription_info", "template", @@ -11286,6 +11384,7 @@ "EventJournalWriter", "EventSink", "Executor", + "FailureChunk", "FileNotFoundError", "FilePart", "FinalOutputPolicyRequest", @@ -11365,6 +11464,7 @@ "SessionWarningPayload", "StatusEventPayload", "StreamChunk", + "StreamFailure", "StreamOptions", "SubscriptionInfo", "Template", @@ -11716,6 +11816,13 @@ "source": "Sources/PromptyModel/contracts/error_event_payload.swift", "protocol": false }, + { + "name": "FailureChunk", + "kind": "value", + "group": "contracts", + "source": "Sources/PromptyModel/contracts/stream_chunk.swift", + "protocol": false + }, { "name": "FileNotFoundError", "kind": "value", @@ -12220,6 +12327,13 @@ "source": "Sources/PromptyModel/contracts/stream_chunk.swift", "protocol": false }, + { + "name": "StreamFailure", + "kind": "value", + "group": "contracts", + "source": "Sources/PromptyModel/contracts/stream_failure.swift", + "protocol": false + }, { "name": "StreamOptions", "kind": "value", @@ -12593,6 +12707,7 @@ "EnginePermissionDecision", "ErrorChunk", "ErrorEventPayload", + "FailureChunk", "FileNotFoundError", "FilePart", "FinalOutputPolicyRequest", @@ -12665,6 +12780,7 @@ "SessionWarningPayload", "StatusEventPayload", "StreamChunk", + "StreamFailure", "StreamOptions", "SubscriptionInfo", "Template", @@ -12786,6 +12902,7 @@ "session_warning_payload.swift", "status_event_payload.swift", "stream_chunk.swift", + "stream_failure.swift", "stream_options.swift", "subscription_info.swift", "template.swift", @@ -13464,6 +13581,7 @@ "Sources/PromptyModel/contracts/session_warning_payload.swift", "Sources/PromptyModel/contracts/status_event_payload.swift", "Sources/PromptyModel/contracts/stream_chunk.swift", + "Sources/PromptyModel/contracts/stream_failure.swift", "Sources/PromptyModel/contracts/stream_options.swift", "Sources/PromptyModel/contracts/subscription_info.swift", "Sources/PromptyModel/contracts/template.swift", @@ -13576,6 +13694,7 @@ "EventJournalWriter", "EventSink", "Executor", + "FailureChunk", "FileNotFoundError", "FilePart", "FinalOutputPolicyRequest", @@ -13655,6 +13774,7 @@ "SessionWarningPayload", "StatusEventPayload", "StreamChunk", + "StreamFailure", "StreamOptions", "SubscriptionInfo", "Template", @@ -14006,6 +14126,13 @@ "source": "./contracts/error-event-payload", "protocol": false }, + { + "name": "FailureChunk", + "kind": "value", + "group": "contracts", + "source": "./contracts/stream-chunk", + "protocol": false + }, { "name": "FileNotFoundError", "kind": "value", @@ -14510,6 +14637,13 @@ "source": "./contracts/stream-chunk", "protocol": false }, + { + "name": "StreamFailure", + "kind": "value", + "group": "contracts", + "source": "./contracts/stream-failure", + "protocol": false + }, { "name": "StreamOptions", "kind": "value", @@ -14883,6 +15017,7 @@ "EnginePermissionDecision", "ErrorChunk", "ErrorEventPayload", + "FailureChunk", "FileNotFoundError", "FilePart", "FinalOutputPolicyRequest", @@ -14955,6 +15090,7 @@ "SessionWarningPayload", "StatusEventPayload", "StreamChunk", + "StreamFailure", "StreamOptions", "SubscriptionInfo", "Template", @@ -15076,6 +15212,7 @@ "session-warning-payload", "status-event-payload", "stream-chunk", + "stream-failure", "stream-options", "subscription-info", "template", @@ -15754,6 +15891,7 @@ "./contracts/session-warning-payload", "./contracts/status-event-payload", "./contracts/stream-chunk", + "./contracts/stream-failure", "./contracts/stream-options", "./contracts/subscription-info", "./contracts/template", diff --git a/schema/tsp-output/.typra-generated/manifest.json b/schema/tsp-output/.typra-generated/manifest.json index 888e61486..cd6ed5e73 100644 --- a/schema/tsp-output/.typra-generated/manifest.json +++ b/schema/tsp-output/.typra-generated/manifest.json @@ -178,6 +178,11 @@ "path": "../runtime/csharp/Prompty.Core.Tests/Model/Contracts/Events/ErrorEventPayloadConversionTests.cs", "marker": true }, + { + "outputRoot": "../runtime/csharp/Prompty.Core.Tests/Model", + "path": "../runtime/csharp/Prompty.Core.Tests/Model/Contracts/Events/FailureChunkConversionTests.cs", + "marker": true + }, { "outputRoot": "../runtime/csharp/Prompty.Core.Tests/Model", "path": "../runtime/csharp/Prompty.Core.Tests/Model/Contracts/Events/HarnessContextConversionTests.cs", @@ -303,6 +308,11 @@ "path": "../runtime/csharp/Prompty.Core.Tests/Model/Contracts/Events/StreamChunkConversionTests.cs", "marker": true }, + { + "outputRoot": "../runtime/csharp/Prompty.Core.Tests/Model", + "path": "../runtime/csharp/Prompty.Core.Tests/Model/Contracts/Events/StreamFailureConversionTests.cs", + "marker": true + }, { "outputRoot": "../runtime/csharp/Prompty.Core.Tests/Model", "path": "../runtime/csharp/Prompty.Core.Tests/Model/Contracts/Events/TextChunkConversionTests.cs", @@ -938,6 +948,11 @@ "path": "../runtime/csharp/Prompty.Core/Model/Contracts/Events/ErrorEventPayload.cs", "marker": true }, + { + "outputRoot": "../runtime/csharp/Prompty.Core/Model", + "path": "../runtime/csharp/Prompty.Core/Model/Contracts/Events/FailureChunk.cs", + "marker": true + }, { "outputRoot": "../runtime/csharp/Prompty.Core/Model", "path": "../runtime/csharp/Prompty.Core/Model/Contracts/Events/HarnessContext.cs", @@ -1093,6 +1108,16 @@ "path": "../runtime/csharp/Prompty.Core/Model/Contracts/Events/StreamChunk.cs", "marker": true }, + { + "outputRoot": "../runtime/csharp/Prompty.Core/Model", + "path": "../runtime/csharp/Prompty.Core/Model/Contracts/Events/StreamFailure.cs", + "marker": true + }, + { + "outputRoot": "../runtime/csharp/Prompty.Core/Model", + "path": "../runtime/csharp/Prompty.Core/Model/Contracts/Events/StreamFailureOutcome.cs", + "marker": true + }, { "outputRoot": "../runtime/csharp/Prompty.Core/Model", "path": "../runtime/csharp/Prompty.Core/Model/Contracts/Events/TextChunk.cs", @@ -2063,6 +2088,11 @@ "path": "../runtime/go/prompty/model/executor.go", "marker": true }, + { + "outputRoot": "../runtime/go/prompty/model", + "path": "../runtime/go/prompty/model/failure_chunk_test.go", + "marker": true + }, { "outputRoot": "../runtime/go/prompty/model", "path": "../runtime/go/prompty/model/file_not_found_error_test.go", @@ -2763,6 +2793,16 @@ "path": "../runtime/go/prompty/model/stream_chunk.go", "marker": true }, + { + "outputRoot": "../runtime/go/prompty/model", + "path": "../runtime/go/prompty/model/stream_failure_test.go", + "marker": true + }, + { + "outputRoot": "../runtime/go/prompty/model", + "path": "../runtime/go/prompty/model/stream_failure.go", + "marker": true + }, { "outputRoot": "../runtime/go/prompty/model", "path": "../runtime/go/prompty/model/stream_options_test.go", @@ -3488,6 +3528,11 @@ "path": "../runtime/python/prompty/prompty/model/contracts/events/_StreamChunk.py", "marker": true }, + { + "outputRoot": "../runtime/python/prompty/prompty/model", + "path": "../runtime/python/prompty/prompty/model/contracts/events/_StreamFailure.py", + "marker": true + }, { "outputRoot": "../runtime/python/prompty/prompty/model", "path": "../runtime/python/prompty/prompty/model/contracts/events/_ThinkingEventPayload.py", @@ -4103,6 +4148,11 @@ "path": "../runtime/python/prompty/tests/model/contracts/events/test_error_event_payload.py", "marker": true }, + { + "outputRoot": "../runtime/python/prompty/tests/model", + "path": "../runtime/python/prompty/tests/model/contracts/events/test_failure_chunk.py", + "marker": true + }, { "outputRoot": "../runtime/python/prompty/tests/model", "path": "../runtime/python/prompty/tests/model/contracts/events/test_harness_context.py", @@ -4228,6 +4278,11 @@ "path": "../runtime/python/prompty/tests/model/contracts/events/test_stream_chunk.py", "marker": true }, + { + "outputRoot": "../runtime/python/prompty/tests/model", + "path": "../runtime/python/prompty/tests/model/contracts/events/test_stream_failure.py", + "marker": true + }, { "outputRoot": "../runtime/python/prompty/tests/model", "path": "../runtime/python/prompty/tests/model/contracts/events/test_text_chunk.py", @@ -4938,6 +4993,11 @@ "path": "../runtime/rust/prompty/src/model/contracts/events/stream_chunk.rs", "marker": true }, + { + "outputRoot": "../runtime/rust/prompty/src/model", + "path": "../runtime/rust/prompty/src/model/contracts/events/stream_failure.rs", + "marker": true + }, { "outputRoot": "../runtime/rust/prompty/src/model", "path": "../runtime/rust/prompty/src/model/contracts/events/thinking_event_payload.rs", @@ -5723,6 +5783,11 @@ "path": "../runtime/rust/prompty/tests/model/contracts/events/stream_chunk_test.rs", "marker": true }, + { + "outputRoot": "../runtime/rust/prompty/tests/model", + "path": "../runtime/rust/prompty/tests/model/contracts/events/stream_failure_test.rs", + "marker": true + }, { "outputRoot": "../runtime/rust/prompty/tests/model", "path": "../runtime/rust/prompty/tests/model/contracts/events/thinking_event_payload_test.rs", @@ -6413,6 +6478,11 @@ "path": "../runtime/swift/prompty-model/Sources/PromptyModel/Contracts/Events/stream_chunk.swift", "marker": true }, + { + "outputRoot": "../runtime/swift/prompty-model", + "path": "../runtime/swift/prompty-model/Sources/PromptyModel/Contracts/Events/stream_failure.swift", + "marker": true + }, { "outputRoot": "../runtime/swift/prompty-model", "path": "../runtime/swift/prompty-model/Sources/PromptyModel/Contracts/Events/thinking_event_payload.swift", @@ -7128,6 +7198,11 @@ "path": "../runtime/swift/prompty-model/Tests/PromptyModelTests/Contracts/Events/StreamChunkTests.swift", "marker": true }, + { + "outputRoot": "../runtime/swift/prompty-model", + "path": "../runtime/swift/prompty-model/Tests/PromptyModelTests/Contracts/Events/StreamFailureTests.swift", + "marker": true + }, { "outputRoot": "../runtime/swift/prompty-model", "path": "../runtime/swift/prompty-model/Tests/PromptyModelTests/Contracts/Events/ThinkingEventPayloadTests.swift", @@ -7783,6 +7858,11 @@ "path": "../runtime/typescript/packages/core/src/model/contracts/events/stream-chunk.ts", "marker": true }, + { + "outputRoot": "../runtime/typescript/packages/core/src/model", + "path": "../runtime/typescript/packages/core/src/model/contracts/events/stream-failure.ts", + "marker": true + }, { "outputRoot": "../runtime/typescript/packages/core/src/model", "path": "../runtime/typescript/packages/core/src/model/contracts/events/thinking-event-payload.ts", @@ -8483,6 +8563,11 @@ "path": "../runtime/typescript/packages/core/tests/model/contracts/events/error-event-payload.test.ts", "marker": true }, + { + "outputRoot": "../runtime/typescript/packages/core/tests/model", + "path": "../runtime/typescript/packages/core/tests/model/contracts/events/failure-chunk.test.ts", + "marker": true + }, { "outputRoot": "../runtime/typescript/packages/core/tests/model", "path": "../runtime/typescript/packages/core/tests/model/contracts/events/harness-context.test.ts", @@ -8608,6 +8693,11 @@ "path": "../runtime/typescript/packages/core/tests/model/contracts/events/stream-chunk.test.ts", "marker": true }, + { + "outputRoot": "../runtime/typescript/packages/core/tests/model", + "path": "../runtime/typescript/packages/core/tests/model/contracts/events/stream-failure.test.ts", + "marker": true + }, { "outputRoot": "../runtime/typescript/packages/core/tests/model", "path": "../runtime/typescript/packages/core/tests/model/contracts/events/text-chunk.test.ts", @@ -9273,6 +9363,11 @@ "path": "../web/src/content/docs/reference/Executor.md", "marker": true }, + { + "outputRoot": "../web/src/content/docs/reference", + "path": "../web/src/content/docs/reference/FailureChunk.md", + "marker": true + }, { "outputRoot": "../web/src/content/docs/reference", "path": "../web/src/content/docs/reference/FileNotFoundError.md", @@ -9673,6 +9768,11 @@ "path": "../web/src/content/docs/reference/StreamChunk.md", "marker": true }, + { + "outputRoot": "../web/src/content/docs/reference", + "path": "../web/src/content/docs/reference/StreamFailure.md", + "marker": true + }, { "outputRoot": "../web/src/content/docs/reference", "path": "../web/src/content/docs/reference/StreamOptions.md", diff --git a/spec/vectors/process/stream_failure_vectors.json b/spec/vectors/process/stream_failure_vectors.json new file mode 100644 index 000000000..b5777dad8 --- /dev/null +++ b/spec/vectors/process/stream_failure_vectors.json @@ -0,0 +1,80 @@ +[ + { + "name": "stream_refusal_is_determinate", + "description": "A model refusal is a determinate terminal failure and never commits a completion", + "input": { + "provider": "openai", + "events": [ + { + "kind": "provider", + "value": { + "choices": [ + { + "delta": { + "refusal": "I cannot help with that" + } + } + ] + } + } + ] + }, + "expected": { + "chunks": [ + { + "kind": "failure", + "failure": { + "outcome": "determinate", + "message": "Model refused: I cannot help with that" + } + } + ], + "partialText": "", + "requiresReconciliation": false, + "completionCommitted": false + } + }, + { + "name": "partial_text_then_indeterminate_failure", + "description": "A post-open transport failure preserves partial text, requires reconciliation, and never commits a completion", + "input": { + "provider": "openai", + "events": [ + { + "kind": "provider", + "value": { + "choices": [ + { + "delta": { + "content": "partial" + } + } + ] + } + }, + { + "kind": "transportError", + "message": "SSE stream error: connection reset" + } + ] + }, + "expected": { + "chunks": [ + { + "kind": "text", + "value": "partial" + }, + { + "kind": "failure", + "failure": { + "outcome": "indeterminate", + "message": "SSE stream error: connection reset" + } + } + ], + "partialText": "partial", + "requiresReconciliation": true, + "completionCommitted": false + } + } +] diff --git a/vscode/prompty/schemas/FailureChunk.yaml b/vscode/prompty/schemas/FailureChunk.yaml new file mode 100644 index 000000000..7afd7e6a2 --- /dev/null +++ b/vscode/prompty/schemas/FailureChunk.yaml @@ -0,0 +1,17 @@ +$schema: https://json-schema.org/draft/2020-12/schema +$id: FailureChunk.yaml +type: object +properties: + kind: + type: string + const: failure + description: The kind identifier for classified failure chunks + failure: + $ref: StreamFailure.yaml + description: The classified stream failure +required: + - kind + - failure +allOf: + - $ref: StreamChunk.yaml +description: A classified failure chunk from the LLM response stream. diff --git a/vscode/prompty/schemas/StreamChunk.yaml b/vscode/prompty/schemas/StreamChunk.yaml index f88cf9675..bb9d0676c 100644 --- a/vscode/prompty/schemas/StreamChunk.yaml +++ b/vscode/prompty/schemas/StreamChunk.yaml @@ -14,6 +14,8 @@ properties: const: usage - type: string const: error + - type: string + const: failure description: The kind of stream chunk required: - kind diff --git a/vscode/prompty/schemas/StreamFailure.yaml b/vscode/prompty/schemas/StreamFailure.yaml new file mode 100644 index 000000000..4b326ac50 --- /dev/null +++ b/vscode/prompty/schemas/StreamFailure.yaml @@ -0,0 +1,18 @@ +$schema: https://json-schema.org/draft/2020-12/schema +$id: StreamFailure.yaml +type: object +properties: + outcome: + anyOf: + - type: string + const: determinate + - type: string + const: indeterminate + description: Whether the provider outcome is known or requires reconciliation + message: + type: string + description: The human-readable failure message +required: + - outcome + - message +description: A classified terminal failure from an LLM response stream. diff --git a/web/src/content/docs/reference/FailureChunk.md b/web/src/content/docs/reference/FailureChunk.md new file mode 100644 index 000000000..522b641ec --- /dev/null +++ b/web/src/content/docs/reference/FailureChunk.md @@ -0,0 +1,57 @@ +--- +title: "FailureChunk" +description: "Documentation for the FailureChunk type." +slug: "reference/failurechunk" +--- + + +A classified failure chunk from the LLM response stream. + +## Class Diagram + +```mermaid +--- +title: FailureChunk +config: + look: handDrawn + theme: colorful + class: + hideEmptyMembersBox: true +--- +classDiagram + class StreamChunk { + <> + +string kind + } + StreamChunk <|-- FailureChunk + class FailureChunk { + +string kind + +StreamFailure failure + } + class StreamFailure { + +string outcome + +string message + } + FailureChunk *-- StreamFailure +``` + +## Yaml Example + +```yaml +failure: + outcome: indeterminate + message: "SSE stream error: connection reset" +``` + +## Properties + +| Name | Type | Description | +| ---- | ---- | ----------- | +| kind | string | The kind identifier for classified failure chunks | +| failure | [StreamFailure](../streamfailure/) | The classified stream failure | + +## Composed Types + +The following types are composed within `FailureChunk`: + +- [StreamFailure](../streamfailure/) diff --git a/web/src/content/docs/reference/StreamChunk.md b/web/src/content/docs/reference/StreamChunk.md index d866f79ad..72f3bcbbb 100644 --- a/web/src/content/docs/reference/StreamChunk.md +++ b/web/src/content/docs/reference/StreamChunk.md @@ -49,6 +49,11 @@ classDiagram +string message } StreamChunk <|-- ErrorChunk + class FailureChunk { + +string kind + +StreamFailure failure + } + StreamChunk <|-- FailureChunk ``` ## Properties @@ -66,3 +71,4 @@ The following types extend `StreamChunk`: - [ToolChunk](../toolchunk/) - [UsageChunk](../usagechunk/) - [ErrorChunk](../errorchunk/) +- [FailureChunk](../failurechunk/) diff --git a/web/src/content/docs/reference/StreamFailure.md b/web/src/content/docs/reference/StreamFailure.md new file mode 100644 index 000000000..45e925956 --- /dev/null +++ b/web/src/content/docs/reference/StreamFailure.md @@ -0,0 +1,40 @@ +--- +title: "StreamFailure" +description: "Documentation for the StreamFailure type." +slug: "reference/streamfailure" +--- + + +A classified terminal failure from an LLM response stream. + +## Class Diagram + +```mermaid +--- +title: StreamFailure +config: + look: handDrawn + theme: colorful + class: + hideEmptyMembersBox: true +--- +classDiagram + class StreamFailure { + +string outcome + +string message + } +``` + +## Yaml Example + +```yaml +outcome: indeterminate +message: "SSE stream error: connection reset" +``` + +## Properties + +| Name | Type | Description | +| ---- | ---- | ----------- | +| outcome | string | Whether the provider outcome is known or requires reconciliation | +| message | string | The human-readable failure message |