Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions .changesets/1792168841-0385b1d7.yaml
Original file line number Diff line number Diff line change
@@ -0,0 +1,11 @@
id: 1792168841-0385b1d7
features:
- serverEvents
targets:
- go
- cli
type: feat
bump: patch
description: expose the raw event behind each streamed Value through EventStream.RawValue
author: 2ynn
date: "2026-10-16"
10 changes: 10 additions & 0 deletions .changesets/1792168842-05c0f7ca.yaml
Original file line number Diff line number Diff line change
@@ -0,0 +1,10 @@
id: 1792168842-05c0f7ca
features:
- serverEvents
targets:
- cli
type: fix
bump: patch
description: render streamed events from the frame the server sent, so an absent required union no longer aborts the stream and absent, null or undeclared fields are shown as sent
author: 2ynn
date: "2026-10-16"
4 changes: 3 additions & 1 deletion internal/features/tests.go
Original file line number Diff line number Diff line change
Expand Up @@ -499,6 +499,9 @@ const (
TestEventStreamMalformedFrameStrict
TestEventStreamMalformedFrameLenient
TestEventStreamTruncatedBeforeSentinel
TestEventStreamAbsentRequiredUnion
TestEventStreamNullAndUndeclaredFields
TestEventStreamEquivalentNumberLexemes
TestTestHooks
TestTestHooksError
TestTestHooksBeforeCreateRequest
Expand Down Expand Up @@ -695,7 +698,6 @@ const (
TestErrorTaxonomyPrettyResidualDetails
TestErrorTaxonomyReservedKeysAndRenderedClassification
TestEventStreamConnectionRefusedStaysConnectionError
TestEventStreamTruncatedBeforeSentinelIsConnectionError
TestEventStreamDecodeFailureIsProtocolError
TestEventStreamDecodeFailureWithoutJsonErrorType
TestEventStreamErrorEventAgentMode
Expand Down
20 changes: 15 additions & 5 deletions internal/features/tests_generated.go

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 1 addition & 0 deletions templates/perms.go
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,7 @@ var fileMode = map[string]fs.FileMode{
"templates/cli/auxiliary/internal/output/pretty.go.stmpl": 0o644,
"templates/cli/auxiliary/internal/output/progress.go.stmpl": 0o644,
"templates/cli/auxiliary/internal/output/streaming.go.stmpl": 0o644,
"templates/cli/auxiliary/internal/output/streaming_test.go.stmpl": 0o644,
"templates/cli/auxiliary/internal/output/streamproject.go.stmpl": 0o644,
"templates/cli/auxiliary/internal/testclient": 0o20000000755,
"templates/cli/auxiliary/internal/testclient/testclient.go.stmpl": 0o644,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -159,15 +159,21 @@ func normalizeForOutput(content interface{}) (interface{}, error) {
if err != nil {
return nil, err
}
dec := json.NewDecoder(bytes.NewReader(data))
dec.UseNumber()
var parsed interface{}
if err := dec.Decode(&parsed); err != nil {
parsed, err := decodeWithNumbers(data)
if err != nil {
return nil, fmt.Errorf("failed to decode marshalled response: %w", err)
}
return parsed, nil
}

func decodeWithNumbers(data []byte) (interface{}, error) {
dec := json.NewDecoder(bytes.NewReader(data))
dec.UseNumber()
var parsed interface{}
err := dec.Decode(&parsed)
return parsed, err
}

func convertNumbers(v interface{}, conv func(json.Number) interface{}) interface{} {
switch val := v.(type) {
case json.Number:
Expand Down
153 changes: 148 additions & 5 deletions templates/templates/cli/auxiliary/internal/output/streaming.go.stmpl
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ package output
import (
"encoding/json"
"errors"
"math/big"
"reflect"
"strings"

Expand Down Expand Up @@ -60,6 +61,7 @@ func StreamResult(cmd *cobra.Command, res interface{}, streamFieldName string) e
valueMethod := streamVal.MethodByName("Value")
errMethod := streamVal.MethodByName("Err")
closeMethod := streamVal.MethodByName("Close")
rawMethod := streamVal.MethodByName("RawValue")
if !nextMethod.IsValid() || !valueMethod.IsValid() {
// Fallback: not a streaming type — output as regular result
return Result(cmd, res)
Expand Down Expand Up @@ -130,12 +132,21 @@ func StreamResult(cmd *cobra.Command, res interface{}, streamFieldName string) e
}
}

var raw []byte
if rawMethod.IsValid() {
raw = rawMethod.Call(nil)[0].Bytes()
}

if len(valueResults) == 1 {
if body, isErr := streamErrorEvent(item); isErr {
if body, isErr := streamErrorEvent(item, raw); isErr {
return renderAfterFinish(&StreamEventError{Body: string(body)})
}
}

if raw != nil {
item = wireStreamItem(item, raw, format == "table" && jqExpr == "")
}

if projector != nil {
if err := projector.emit(out, item); err != nil {
return finishJoin(err)
Expand Down Expand Up @@ -166,16 +177,148 @@ func StreamResult(cmd *cobra.Command, res interface{}, streamFieldName string) e
return nil
}

func streamErrorEvent(item interface{}) (json.RawMessage, bool) {
type wireEvent struct {
data json.RawMessage
}

func (w wireEvent) MarshalJSON() ([]byte, error) {
return w.data, nil
}

// Re-encoding the decoded model drops undeclared fields and zero-fills absent ones.
func wireStreamItem(item interface{}, raw []byte, table bool) interface{} {
keys, ok := envelopeKeys(reflect.ValueOf(item))
if len(raw) == 0 || !ok {
return item
}
typed, err := normalizeForOutput(item)
// Tables lay out the model's fields and sections, which the wire map loses.
if err == nil && table {
return item
}
frame, err := decodeWithNumbers(raw)
wire, isObject := frame.(map[string]interface{})
if err != nil || !isObject {
return item
}
for name := range wire {
if keys != nil && !keys[name] {
delete(wire, name)
}
}
data, err := json.Marshal(typedLeaves(wire, typed))
if err != nil {
return item
}
return wireEvent{data: data}
}

// nil keys keep the whole frame: a union event that matched no member.
func envelopeKeys(v reflect.Value) (map[string]bool, bool) {
v = derefValue(v)
if !v.IsValid() || v.Kind() != reflect.Struct {
return nil, false
}
for isUnionStruct(v.Type()) {
member := reflect.Value{}
for i := 0; i < v.NumField(); i++ {
if v.Type().Field(i).Tag.Get("union") == "member" && !v.Field(i).IsZero() {
member = derefValue(v.Field(i))
break
}
}
if !member.IsValid() || member.Kind() != reflect.Struct {
return nil, true
}
v = member
}
t := v.Type()
keys := map[string]bool{}
for i := 0; i < t.NumField(); i++ {
f := t.Field(i)
if f.Tag.Get("additionalProperties") == "true" {
return nil, true
}
name := strings.Split(f.Tag.Get("json"), ",")[0]
if name == "-" || (name == "" && !f.IsExported()) {
continue
}
if name == "" {
name = f.Name
}
keys[name] = true
}
return keys, true
}

func typedLeaves(wire, typed interface{}) interface{} {
switch w := wire.(type) {
case nil:
return nil
case map[string]interface{}:
t, _ := typed.(map[string]interface{})
for k, v := range w {
w[k] = typedLeaves(v, t[k])
}
return w
case []interface{}:
t, _ := typed.([]interface{})
for i, v := range w {
var tv interface{}
if i < len(t) {
tv = t[i]
}
w[i] = typedLeaves(v, tv)
}
return w
}
switch typed.(type) {
case nil, map[string]interface{}, []interface{}:
if n, ok := wire.(json.Number); ok {
return canonicalNumber(n)
}
return wire
}
return typed
}

// Bigint and long-decimal lexemes are kept: float64 would change their value.
func canonicalNumber(n json.Number) interface{} {
exact, ok := new(big.Rat).SetString(n.String())
if !ok {
return n
}
f, err := n.Float64()
if err != nil {
return n
}
shortest, err := json.Marshal(f)
if err != nil {
return n
}
if r, ok := new(big.Rat).SetString(string(shortest)); !ok || r.Cmp(exact) != 0 {
return n
}
return json.Number(shortest)
}

func streamErrorEvent(item interface{}, raw []byte) (json.RawMessage, bool) {
u, ok := errorUnionValue(reflect.ValueOf(item))
if !ok {
return nil, false
}
data, err := marshalJSON(u.Interface())
if err != nil {
if data, err := marshalJSON(u.Interface()); err == nil {
return data, true
}
if isUnionStruct(derefValue(reflect.ValueOf(item)).Type()) {
data, err := json.Marshal(wireStreamItem(item, raw, false))
return data, err == nil
}
var frame map[string]json.RawMessage
if err := json.Unmarshal(raw, &frame); err != nil || len(frame["data"]) == 0 {
return nil, false
}
return data, true
return frame["data"], true
}

func errorUnionValue(v reflect.Value) (reflect.Value, bool) {
Expand Down
Loading
Loading