diff --git a/.changesets/1792717600-4c1d2e7a.yaml b/.changesets/1792717600-4c1d2e7a.yaml new file mode 100644 index 000000000..a8297ebaf --- /dev/null +++ b/.changesets/1792717600-4c1d2e7a.yaml @@ -0,0 +1,10 @@ +id: 1792717600-4c1d2e7a +features: + - cliCommands +targets: + - cli +type: feat +bump: minor +description: declare a continuation recipe so a command re-sends its request with resume values from the response while output.outcome classifies the response state as continue, combining the responses (merge arrays, sum numbers) or streaming every response as one output, with --auto-continue and --max-continuations flags and CLI_CONTINUATION_LIMIT when the hop limit is reached +author: TristanSpeakEasy +date: "2026-10-23" diff --git a/internal/extensions/cli_commands.go b/internal/extensions/cli_commands.go index ee4b3ed15..ab73fd6e0 100644 --- a/internal/extensions/cli_commands.go +++ b/internal/extensions/cli_commands.go @@ -6,6 +6,7 @@ import ( "fmt" "math" "regexp" + "slices" "sort" "strconv" "strings" @@ -274,10 +275,46 @@ type CLICommandOutcome struct { // opt-in streamed projection and, when necessary, scalar flags that merge into // a body property shared by every request-body union member. type CLIOperation struct { - OperationID string `json:"operationId" yaml:"operationId"` - Output *CLICommandOutput `json:"output,omitempty" yaml:"output,omitempty"` - Flags []CLICommandInput `json:"flags,omitempty" yaml:"flags,omitempty"` - BodyRequired bool `json:"bodyRequired,omitempty" yaml:"bodyRequired,omitempty"` + OperationID string `json:"operationId" yaml:"operationId"` + Output *CLICommandOutput `json:"output,omitempty" yaml:"output,omitempty"` + Flags []CLICommandInput `json:"flags,omitempty" yaml:"flags,omitempty"` + Continuation *CLICommandContinuation `json:"continuation,omitempty" yaml:"continuation,omitempty"` + BodyRequired bool `json:"bodyRequired,omitempty" yaml:"bodyRequired,omitempty"` +} + +// CLICommandContinuationBinding copies one resume value from the response +// that stopped into the next request. From addresses the value in a discrete +// JSON response and StreamFrom in streamed events (the last present value +// wins); To is the request-body path that receives it. An optional binding +// is skipped when the value is absent, leaving the original request value. +// Flag is the command's declared flag bound to the same body path, resolved +// during linking for resume hints. +type CLICommandContinuationBinding struct { + From string `json:"from,omitempty" yaml:"from,omitempty"` + Pointer string `json:"pointer,omitempty" yaml:"pointer,omitempty"` + StreamFrom []string `json:"streamFrom,omitempty" yaml:"streamFrom,omitempty"` + StreamPointers []string `json:"streamPointers,omitempty" yaml:"streamPointers,omitempty"` + To string `json:"to" yaml:"to"` + ToPointer string `json:"toPointer" yaml:"toPointer"` + ToPath []string `json:"toPath" yaml:"toPath"` + Optional bool `json:"optional,omitempty" yaml:"optional,omitempty"` + Flag string `json:"flag,omitempty" yaml:"flag,omitempty"` +} + +// CLICommandContinuation re-sends a command's request while the response's +// output.outcome state classifies as continue, filling Bind values from the +// response that stopped. Merge paths are arrays concatenated across hops and +// Sum paths numbers summed across hops; every other field of a discrete +// result comes from the last hop. +type CLICommandContinuation struct { + Bind []CLICommandContinuationBinding `json:"bind" yaml:"bind"` + MergeFrom []string `json:"mergeFrom,omitempty" yaml:"mergeFrom,omitempty"` + Merge []string `json:"merge,omitempty" yaml:"merge,omitempty"` + SumFrom []string `json:"sumFrom,omitempty" yaml:"sumFrom,omitempty"` + Sum []string `json:"sum,omitempty" yaml:"sum,omitempty"` + MaxHops int `json:"maxHops" yaml:"maxHops"` + Default bool `json:"default" yaml:"default"` + OperationID string `json:"operationId,omitempty" yaml:"operationId,omitempty"` } // CLICommandAsyncParameter identifies the poll-operation parameter that @@ -382,6 +419,7 @@ type CLICommand struct { Presets []CLICommandPreset `json:"presets" yaml:"presets"` CatalogDefaults []CLICatalogDefault `json:"catalogDefaults,omitempty" yaml:"catalogDefaults,omitempty"` Async *CLICommandAsync `json:"async,omitempty" yaml:"async,omitempty"` + Continuation *CLICommandContinuation `json:"continuation,omitempty" yaml:"continuation,omitempty"` Output *CLICommandOutput `json:"output" yaml:"output"` Examples []CLICommandExample `json:"examples" yaml:"examples"` Help *CLICommandHelp `json:"help,omitempty" yaml:"help,omitempty"` @@ -465,6 +503,9 @@ var cliReservedFlagNames = map[string]bool{ "resume": true, "poll-interval": true, "poll-timeout": true, + // Owned by continuation commands, reserved for the same reason. + "auto-continue": true, + "max-continuations": true, } // cliReservedFlagShorthands are owned by the generated runtime's persistent @@ -534,6 +575,7 @@ var CLIRuntimeHintReasons = []string{ "CLI_ASYNC_TIMEOUT", "CLI_ASYNC_UNKNOWN_STATE", "CLI_OUTCOME_FAILED", + "CLI_CONTINUATION_LIMIT", } // cliReservedCommandKeys are keys reserved for future capabilities. Each maps @@ -549,7 +591,7 @@ var cliReservedCommandKeys = map[string]string{ var cliCommandKeys = []string{ "category", "summary", "tagline", "description", "op", "routes", "planned", "custom", "group", "source", "override", - "args", "flags", "preset", "hints", "examples", "help", "jq", "output", "async", + "args", "flags", "preset", "hints", "examples", "help", "jq", "output", "async", "continuation", } var cliOutputKeys = []string{"artifact", "stream", "outcome"} @@ -560,17 +602,30 @@ var cliOutputStreamKeys = []string{"select", "metadata"} var cliOutputStreamMetadataKeys = []string{"select", "label", "complete", "when"} -var cliOperationKeys = []string{"output", "flags"} +var cliOperationKeys = []string{"output", "flags", "continuation"} + +var ( + cliContinuationKeys = []string{"bind", "merge", "sum", "maxHops", "default"} + cliContinuationBindingKeys = []string{"from", "streamFrom", "to", "optional"} + // cliOutcomeStateKind extends the async classifications with continue, + // which only a continuation recipe acts on; async polling has no resume + // value to send. + cliOutcomeStateKind = []string{"pending", "success", "failure", "handoff", "continue"} +) + +const cliContinuationDefaultMaxHops = 32 // These flags are conditionally registered by operation commands. Operation // declarations reserve them even when a particular operation does not use the // corresponding feature, so adding pagination or binary output later cannot // turn a valid manifest into a pflag registration panic. var cliOperationReservedFlagNames = map[string]string{ - "all": "pagination", - "max-pages": "pagination", - "output-file": "binary response output", - "output-b64": "binary response output", + "all": "pagination", + "max-pages": "pagination", + "output-file": "binary response output", + "output-b64": "binary response output", + "auto-continue": "continuation", + "max-continuations": "continuation", } var cliInputKeys = []string{ @@ -858,12 +913,16 @@ func (d *cliManifestDecoder) decodeOperation(opID string, node *yaml.Node) (*CLI return nil, err } op.Flags = flags + case "continuation": + if op.Continuation, err = d.decodeContinuation(fmt.Sprintf("operation %q", opID), entry.Value); err != nil { + return nil, err + } default: return nil, fmt.Errorf("line %d: operation %q has unknown key %q%s", entry.Key.Line, opID, entry.Key.Value, cliDidYouMean(entry.Key.Value, cliOperationKeys)) } } - if op.Output == nil && len(op.Flags) == 0 { - return nil, fmt.Errorf("line %d: operation %q must declare output or flags", node.Line, opID) + if op.Output == nil && len(op.Flags) == 0 && op.Continuation == nil { + return nil, fmt.Errorf("line %d: operation %q must declare output, flags, or continuation", node.Line, opID) } return op, nil } @@ -921,7 +980,7 @@ func (d *cliManifestDecoder) decodeCommand(key string, path []string, id string, var ( opNode, routesNode, plannedNode, customNode, groupNode, sourceNode *yaml.Node argsNode, flagsNode, presetNode, hintsNode, exampNode, helpNode *yaml.Node - outputNode, asyncNode, overrideNode *yaml.Node + outputNode, asyncNode, overrideNode, continuationNode *yaml.Node ) for _, entry := range entries { @@ -988,6 +1047,8 @@ func (d *cliManifestDecoder) decodeCommand(key string, path []string, id string, outputNode = entry.Value case "async": asyncNode = entry.Value + case "continuation": + continuationNode = entry.Value default: return nil, fmt.Errorf("line %d: command %q has unknown key %q%s", entry.Key.Line, key, keyName, cliDidYouMean(keyName, cliCommandKeys)) } @@ -1022,6 +1083,14 @@ func (d *cliManifestDecoder) decodeCommand(key string, path []string, id string, return nil, fmt.Errorf("command %q declares both async and output.outcome; async.states already classifies the terminal poll state", key) } } + if continuationNode != nil { + if cmd.Continuation, err = d.decodeContinuation(fmt.Sprintf("command %q", key), continuationNode); err != nil { + return nil, err + } + if cmd.Async != nil { + return nil, fmt.Errorf("command %q declares both async and continuation; a continuation re-sends the command's own request and cannot follow a poll", key) + } + } if err := d.decodeSource(cmd, key, node, opNode, routesNode, plannedNode, customNode, groupNode, sourceNode); err != nil { return nil, err @@ -1029,6 +1098,9 @@ func (d *cliManifestDecoder) decodeCommand(key string, path []string, id string, if overrideNode != nil && routesNode == nil { return nil, fmt.Errorf("line %d: command %q: override is only valid with a routes: dispatch map", overrideNode.Line, key) } + if continuationNode != nil && len(cmd.Source.Routes) > 1 { + return nil, fmt.Errorf("line %d: command %q: continuation requires a single-operation command; dispatch commands with several routes cannot declare it", continuationNode.Line, key) + } // Group tags categorize top-level command groups (help categories are // root-level sections); the restriction applies to both the group: true @@ -1055,6 +1127,9 @@ func (d *cliManifestDecoder) decodeCommand(key string, path []string, id string, if asyncNode != nil { illegal = append(illegal, "async") } + if continuationNode != nil { + illegal = append(illegal, "continuation") + } if cmd.Source.Type == CLICommandSourceGroup && exampNode != nil { illegal = append(illegal, "examples") } @@ -1494,14 +1569,141 @@ func (d *cliManifestDecoder) decodeOutcomeStates(owner string, node *yaml.Node, if err != nil { return err } - if !cliContains(cliAsyncStateKind, class) { - return fmt.Errorf("line %d: %s output.outcome.states[%q] has unsupported classification %q (expected %s)", entry.Value.Line, owner, state, class, cliCandidateList(cliAsyncStateKind)) + if !cliContains(cliOutcomeStateKind, class) { + return fmt.Errorf("line %d: %s output.outcome.states[%q] has unsupported classification %q (expected %s)", entry.Value.Line, owner, state, class, cliCandidateList(cliOutcomeStateKind)) } states[state] = class } return nil } +// decodeContinuation decodes a continuation recipe. Pointer resolvability, +// the request-body targets, and the required output.outcome continue state +// are checked during linking. +func (d *cliManifestDecoder) decodeContinuation(owner string, node *yaml.Node) (*CLICommandContinuation, error) { + entries, err := cliMapEntries(node, owner+" continuation") + if err != nil { + return nil, err + } + recipe := &CLICommandContinuation{MaxHops: cliContinuationDefaultMaxHops, Default: true} + var bindNode *yaml.Node + for _, entry := range entries { + switch entry.Key.Value { + case "bind": + bindNode = entry.Value + case "merge": + recipe.MergeFrom, recipe.Merge, err = cliDecodeContinuationPaths(entry.Value, "continuation.merge", true) + case "sum": + recipe.SumFrom, recipe.Sum, err = cliDecodeContinuationPaths(entry.Value, "continuation.sum", true) + case "maxHops": + if entry.Value.Kind != yaml.ScalarNode || entry.Value.Tag != "!!int" { + err = stderrors.New("continuation.maxHops must be an integer of at least 1") + break + } + recipe.MaxHops, err = strconv.Atoi(entry.Value.Value) + if err == nil && recipe.MaxHops < 1 { + err = fmt.Errorf("continuation.maxHops must be at least 1, got %d", recipe.MaxHops) + } + case "default": + recipe.Default, err = cliScalarBool(entry.Value, "continuation.default") + default: + return nil, fmt.Errorf("line %d: %s continuation has unknown key %q%s", entry.Key.Line, owner, entry.Key.Value, cliDidYouMean(entry.Key.Value, cliContinuationKeys)) + } + if err != nil { + return nil, fmt.Errorf("line %d: %s: %w", entry.Value.Line, owner, err) + } + } + if bindNode == nil || bindNode.Kind != yaml.SequenceNode || len(bindNode.Content) == 0 { + return nil, fmt.Errorf("line %d: %s continuation requires bind (a non-empty list of {from, to} resume values copied into the next request)", node.Line, owner) + } + targets := map[string]bool{} + required := false + for _, item := range bindNode.Content { + binding, err := d.decodeContinuationBinding(owner, item) + if err != nil { + return nil, err + } + if targets[binding.ToPointer] { + return nil, fmt.Errorf("line %d: %s continuation.bind writes %s more than once", item.Line, owner, binding.To) + } + targets[binding.ToPointer] = true + required = required || !binding.Optional + recipe.Bind = append(recipe.Bind, *binding) + } + if !required { + return nil, fmt.Errorf("line %d: %s continuation.bind must include at least one binding that is not optional; the next request needs a resume value", bindNode.Line, owner) + } + return recipe, nil +} + +func (d *cliManifestDecoder) decodeContinuationBinding(owner string, node *yaml.Node) (*CLICommandContinuationBinding, error) { + entries, err := cliMapEntries(node, owner+" continuation.bind entry") + if err != nil { + return nil, err + } + binding := &CLICommandContinuationBinding{} + for _, entry := range entries { + switch entry.Key.Value { + case "from": + binding.From, binding.Pointer, _, err = cliDecodeAsyncPointer(entry.Value, "continuation.bind.from") + case "streamFrom": + binding.StreamFrom, binding.StreamPointers, err = cliDecodeContinuationPaths(entry.Value, "continuation.bind.streamFrom", false) + case "to": + var segments []CLICommandAsyncPathSegment + binding.To, binding.ToPointer, segments, err = cliDecodeAsyncPointer(entry.Value, "continuation.bind.to") + for _, segment := range segments { + if segment.IsIndex { + err = fmt.Errorf("continuation.bind.to %s must name object properties only", binding.To) + break + } + binding.ToPath = append(binding.ToPath, segment.Field) + } + case "optional": + binding.Optional, err = cliScalarBool(entry.Value, "continuation.bind.optional") + default: + return nil, fmt.Errorf("line %d: %s continuation.bind has unknown key %q%s", entry.Key.Line, owner, entry.Key.Value, cliDidYouMean(entry.Key.Value, cliContinuationBindingKeys)) + } + if err != nil { + return nil, fmt.Errorf("line %d: %s: %w", entry.Value.Line, owner, err) + } + } + if binding.To == "" { + return nil, fmt.Errorf("line %d: %s continuation.bind entry requires to (the request-body path that receives the value)", node.Line, owner) + } + if binding.From == "" && len(binding.StreamFrom) == 0 { + return nil, fmt.Errorf("line %d: %s continuation.bind entry for %s requires from (JSON response) or streamFrom (streamed events)", node.Line, owner, binding.To) + } + return binding, nil +} + +// cliDecodeContinuationPaths accepts one singular JSONPath or a list of them +// and returns the authored paths with their RFC 6901 lowerings. +func cliDecodeContinuationPaths(node *yaml.Node, what string, propertiesOnly bool) ([]string, []string, error) { + items := []*yaml.Node{node} + if node.Kind == yaml.SequenceNode { + if len(node.Content) == 0 { + return nil, nil, fmt.Errorf("%s must list at least one path", what) + } + items = node.Content + } + var authored, pointers []string + for _, item := range items { + from, pointer, segments, err := cliDecodeAsyncPointer(item, what) + if err != nil { + return nil, nil, err + } + if propertiesOnly && slices.ContainsFunc(segments, func(segment CLICommandAsyncPathSegment) bool { return segment.IsIndex }) { + return nil, nil, fmt.Errorf("%s %s must name object properties only", what, from) + } + if cliContains(pointers, pointer) { + return nil, nil, fmt.Errorf("%s lists %s more than once", what, from) + } + authored = append(authored, from) + pointers = append(pointers, pointer) + } + return authored, pointers, nil +} + // decodeArtifact decodes and validates an output.artifact declaration. The // content pointer's resolvability against the response schema is checked // during linking; this stage validates the local grammar. diff --git a/internal/extensions/cli_commands_link.go b/internal/extensions/cli_commands_link.go index 73461ddc9..0a55471a5 100644 --- a/internal/extensions/cli_commands_link.go +++ b/internal/extensions/cli_commands_link.go @@ -134,9 +134,310 @@ func (d *cliManifestDecoder) linkManifest(manifest *CLICommandManifest) error { } } inheritOperationOutcomes(manifest) + inheritOperationContinuations(manifest) + return d.linkContinuations(manifest) +} + +// inheritOperationContinuations gives a single-route, non-async intent +// without its own continuation the recipe its operation entry declared, as +// output.outcome is inherited. Bindings are copied because hint flags are +// resolved against each command's own flags. +func inheritOperationContinuations(manifest *CLICommandManifest) { + recipes := map[string]*CLICommandContinuation{} + for i := range manifest.Operations { + if op := &manifest.Operations[i]; op.Continuation != nil { + recipes[op.OperationID] = op.Continuation + } + } + if len(recipes) == 0 { + return + } + for i := range manifest.Commands { + cmd := &manifest.Commands[i] + if cmd.Source.Type != CLICommandSourceOperation || len(cmd.Source.Routes) != 1 || cmd.Async != nil || cmd.Continuation != nil { + continue + } + recipe, ok := recipes[cmd.Source.Routes[0].OperationID] + if !ok { + continue + } + copied := *recipe + copied.Bind = append([]CLICommandContinuationBinding(nil), recipe.Bind...) + cmd.Continuation = &copied + } +} + +func (d *cliManifestDecoder) linkContinuations(manifest *CLICommandManifest) error { + asyncCreates := map[string]string{} + for i := range manifest.Commands { + cmd := &manifest.Commands[i] + if cmd.Async != nil && len(cmd.Source.Routes) > 0 { + asyncCreates[cmd.Source.Routes[0].OperationID] = strings.Join(cmd.Path, " ") + } + } + for i := range manifest.Operations { + op := &manifest.Operations[i] + if op.Continuation == nil { + continue + } + var outcome *CLICommandOutcome + if op.Output != nil { + outcome = op.Output.Outcome + } + if err := d.linkContinuation(fmt.Sprintf("operation %q", op.OperationID), op.Continuation, op.OperationID, outcome, op.Flags, asyncCreates); err != nil { + return err + } + } + for i := range manifest.Commands { + cmd := &manifest.Commands[i] + var outcome *CLICommandOutcome + if cmd.Output != nil { + outcome = cmd.Output.Outcome + } + owner := fmt.Sprintf("command %q", strings.Join(cmd.Path, " ")) + if cmd.Continuation == nil || cmd.Source.Type != CLICommandSourceOperation { + if err := rejectUnusedContinueStates(owner, outcome, "add continuation (on the command or its operation entry)"); err != nil { + return err + } + continue + } + if err := d.linkContinuation(owner, cmd.Continuation, cmd.Source.Routes[0].OperationID, outcome, cmd.Flags, asyncCreates); err != nil { + return err + } + } + for i := range manifest.Operations { + op := &manifest.Operations[i] + if op.Continuation != nil || op.Output == nil { + continue + } + if err := rejectUnusedContinueStates(fmt.Sprintf("operation %q", op.OperationID), op.Output.Outcome, "an operation entry's outcome also applies to the generated operation command, so a recipe on an intent does not cover it; move continuation to the operation entry (single-route intents inherit it), move output.outcome to the intent,"); err != nil { + return err + } + } + return nil +} + +// rejectUnusedContinueStates refuses a continue classification that no +// continuation recipe acts on: the command would otherwise exit 0 on a +// response that says it is incomplete. +func rejectUnusedContinueStates(owner string, outcome *CLICommandOutcome, remedy string) error { + if outcome == nil { + return nil + } + var states []string + for state, class := range outcome.States { + if class == "continue" { + states = append(states, state) + } + } + if len(states) == 0 { + return nil + } + sort.Strings(states) + return fmt.Errorf("%s classifies output.outcome state %q as continue but declares no continuation recipe; %s or classify the state as pending, success, or failure", owner, states[0], remedy) +} + +func (d *cliManifestDecoder) linkContinuation(owner string, recipe *CLICommandContinuation, opID string, outcome *CLICommandOutcome, flags []CLICommandInput, asyncCreates map[string]string) error { + opInfo, ok := d.schemaIndex.operations[opID] + if !ok { + return fmt.Errorf("%s continuation: operation %q does not exist in the document%s", owner, opID, cliDidYouMean(opID, d.schemaIndex.opIDs)) + } + recipe.OperationID = opID + if asyncCommand, ok := asyncCreates[opID]; ok { + return fmt.Errorf("%s declares continuation on operation %q, which async command %q also uses as its create operation; continuation and async cannot share an operation in v1", owner, opID, asyncCommand) + } + if outcome == nil { + return fmt.Errorf("%s declares continuation but no output.outcome; declare a states table with at least one continue state on the command or its operation entry", owner) + } + hasContinue := false + for _, class := range outcome.States { + hasContinue = hasContinue || class == "continue" + } + if !hasContinue { + return fmt.Errorf("%s continuation requires output.outcome.states to classify at least one state as continue", owner) + } + hasJSON, hasStream := len(opInfo.jsonResponses) > 0, opInfo.streamMediaSeen + if !hasJSON && !hasStream { + return fmt.Errorf("%s continuation: operation %q has no application/json or streaming 2xx response to read resume values from", owner, opID) + } + if hasJSON && outcome.StatePointer == "" { + return fmt.Errorf("%s continuation: operation %q returns application/json, so output.outcome must declare statePointer", owner, opID) + } + if hasStream && outcome.StreamStatePointer == "" { + return fmt.Errorf("%s continuation: operation %q streams its response, so output.outcome must declare streamStatePointer", owner, opID) + } + if opInfo.bodySchema == nil { + return fmt.Errorf("%s continuation: operation %q has no application/json request body to carry resume values", owner, opID) + } + + for i := range recipe.Bind { + binding := &recipe.Bind[i] + if binding.From != "" { + if !hasJSON { + return fmt.Errorf("%s continuation.bind.from %s: operation %q has no application/json 2xx response; use streamFrom for streamed events", owner, binding.From, opID) + } + for _, response := range opInfo.jsonResponses { + if _, err := d.linkAsyncResponsePointer(binding.From, response.schema, false); err != nil { + return fmt.Errorf("%s continuation.bind.from %s in operation %q response %s: %w", owner, binding.From, opID, response.code, err) + } + } + } else if hasJSON && !binding.Optional { + return fmt.Errorf("%s continuation.bind entry for %s requires from: operation %q returns application/json", owner, binding.To, opID) + } + if len(binding.StreamFrom) > 0 { + if !hasStream { + return fmt.Errorf("%s continuation.bind.streamFrom: operation %q has no streaming 2xx response", owner, opID) + } + for _, path := range binding.StreamFrom { + if err := d.linkStreamSelection(owner, "continuation.bind.streamFrom", path, opInfo); err != nil { + return err + } + } + } else if hasStream && !binding.Optional { + return fmt.Errorf("%s continuation.bind entry for %s requires streamFrom: operation %q streams its response", owner, binding.To, opID) + } + if err := d.linkContinuationTarget(owner, opID, binding, opInfo.bodySchema); err != nil { + return err + } + binding.Flag = "" + for _, flag := range flags { + if flag.Bind != nil && flag.Bind.In == "body" && flag.Bind.Pointer == binding.ToPointer { + binding.Flag = flag.Name + break + } + } + } + + if (len(recipe.Merge) > 0 || len(recipe.Sum) > 0) && !hasJSON { + return fmt.Errorf("%s continuation.merge and continuation.sum apply to application/json responses, and operation %q has none", owner, opID) + } + for _, path := range recipe.MergeFrom { + if err := d.linkContinuationResultPath(owner, "continuation.merge", path, opID, opInfo, func(kind string) bool { return kind == "array" }, "an array"); err != nil { + return err + } + } + for _, path := range recipe.SumFrom { + if err := d.linkContinuationResultPath(owner, "continuation.sum", path, opID, opInfo, func(kind string) bool { return kind == "int" || kind == "float" }, "a number"); err != nil { + return err + } + } + return nil +} + +func (d *cliManifestDecoder) linkContinuationResultPath(owner, field, path, opID string, opInfo *cliOperationInfo, accept func(string) bool, expected string) error { + for _, response := range opInfo.jsonResponses { + views, err := d.walkResponsePath(path, response.schema) + if err != nil { + return fmt.Errorf("%s %s %s in operation %q response %s: %w", owner, field, path, opID, response.code, err) + } + facts, err := d.responseLeafFacts(views) + if err != nil { + return fmt.Errorf("%s %s %s in operation %q response %s: %w", owner, field, path, opID, response.code, err) + } + for _, propertyFacts := range facts { + if len(propertyFacts.kinds) != 1 || !accept(propertyFacts.kinds[0]) { + return fmt.Errorf("%s %s %s in operation %q response %s must select %s in every response variant", owner, field, path, opID, response.code, expected) + } + } + } + return nil +} + +// linkContinuationTarget requires the bound request-body path to be a +// writable string property in every request-body variant, so a resume value +// always lands in a field the API accepts. +func (d *cliManifestDecoder) linkContinuationTarget(owner, opID string, binding *CLICommandContinuationBinding, bodySchema any) error { + for i, raw := range d.continuationTargetVariants(bodySchema) { + if err := d.linkContinuationTargetPath(owner, opID, binding, raw, 0, i+1); err != nil { + return err + } + } + return nil +} + +func (d *cliManifestDecoder) linkContinuationTargetPath(owner, opID string, binding *CLICommandContinuationBinding, schema any, depth, bodyVariant int) error { + name := binding.ToPath[depth] + for _, raw := range d.continuationTargetVariants(schema) { + variant, err := d.resolveObjectSchema(raw, nil) + if err != nil { + return fmt.Errorf("%s continuation.bind.to %s: request body variant %d of operation %q: %w", owner, binding.To, bodyVariant, opID, err) + } + prop, exists := variant.properties[name] + if !exists { + return fmt.Errorf("%s continuation.bind.to %s: property %q is absent from request body variant %s of operation %q%s", owner, binding.To, name, cliSchemaLabel(variant), opID, cliDidYouMean(name, variant.propOrder)) + } + if depth < len(binding.ToPath)-1 { + if err := d.linkContinuationTargetPath(owner, opID, binding, prop, depth+1, bodyVariant); err != nil { + return err + } + continue + } + facts, err := d.propertyFacts(prop, nil) + if err != nil { + return fmt.Errorf("%s continuation.bind.to %s in request body variant %s of operation %q: %w", owner, binding.To, cliSchemaLabel(variant), opID, err) + } + readOnly := facts.readOnly + if leaf := d.continuationTargetLeaf(prop); leaf != nil { + facts, err = d.propertyFacts(leaf, nil) + if err != nil { + return fmt.Errorf("%s continuation.bind.to %s in request body variant %s of operation %q: %w", owner, binding.To, cliSchemaLabel(variant), opID, err) + } + readOnly = readOnly || facts.readOnly + } + if readOnly { + return fmt.Errorf("%s continuation.bind.to %s names readOnly property %q in request body variant %s of operation %q", owner, binding.To, name, cliSchemaLabel(variant), opID) + } + if len(facts.kinds) != 1 || facts.kinds[0] != "string" { + return fmt.Errorf("%s continuation.bind.to %s must name a string property in request body variant %s of operation %q", owner, binding.To, cliSchemaLabel(variant), opID) + } + } return nil } +// continuationTargetLeaf drops null from a target's type list or union, so a +// nullable string accepts a resume value; it returns nil when nothing changes. +func (d *cliManifestDecoder) continuationTargetLeaf(prop any) any { + leaf := prop + if variants := d.continuationTargetVariants(prop); len(variants) == 1 { + leaf = variants[0] + } + if leafMap, ok := leaf.(map[string]any); ok { + if stripped := cliWithoutNullType(leafMap); stripped != nil { + leaf = stripped + } + } + if reflect.DeepEqual(leaf, prop) { + return nil + } + return leaf +} + +// continuationTargetVariants expands an object union into its members, each +// conjoined with the keywords the union declares beside oneOf/anyOf. A +// null-only member is skipped: there is no object to write into. +func (d *cliManifestDecoder) continuationTargetVariants(schema any) []any { + schemaMap, _ := schema.(map[string]any) + unionMap, _ := d.unionBodySchema(schemaMap) + members := cliUnionMembers(unionMap) + if len(members) == 0 { + return []any{schema} + } + shared := cliUnionSharedSchema(unionMap) + variants := make([]any, 0, len(members)) + for _, member := range members { + memberMap, _ := member.(map[string]any) + if types := cliSchemaTypes(memberMap); len(types) == 1 && types[0] == "null" { + continue + } + if shared != nil { + variants = append(variants, cliConjoinSchemas(memberMap, shared)) + } else { + variants = append(variants, member) + } + } + return variants +} + // inheritOperationOutcomes gives a single-route intent without its own // output.outcome the table its operation entry declared, so a failure state // classified for the operation command also fails the intent. Async intents @@ -3734,6 +4035,23 @@ func (d *cliManifestDecoder) resolveObjectSchema(schema any, seen []string) (*cl return out, nil } +// cliUnionSharedSchema returns the keywords declared beside a union's +// oneOf/anyOf, which apply to every member; nil when there are none. +func cliUnionSharedSchema(unionMap map[string]any) map[string]any { + shared := map[string]any{} + for key, value := range unionMap { + switch key { + case "oneOf", "anyOf", "discriminator", "description", "title", "example", "examples": + continue + } + shared[key] = value + } + if len(shared) == 0 { + return nil + } + return shared +} + func cliConjoinSchemas(a, b map[string]any) map[string]any { switch { case a == nil: diff --git a/internal/extensions/cli_commands_test.go b/internal/extensions/cli_commands_test.go index 3fbf0ed5c..684700eff 100644 --- a/internal/extensions/cli_commands_test.go +++ b/internal/extensions/cli_commands_test.go @@ -2201,6 +2201,7 @@ commands: CLI_ASYNC_TIMEOUT: async timeout CLI_ASYNC_UNKNOWN_STATE: async unknown state CLI_OUTCOME_FAILED: outcome failed + CLI_CONTINUATION_LIMIT: continuation limit `) require.NoError(t, err) require.Len(t, manifest.Commands, 1) @@ -5728,6 +5729,438 @@ func TestCLICommands_OutcomeStreamStateAllOfIntersectsEnums(t *testing.T) { } } +const cliContinuationTestSpec = `openapi: 3.1.0 +info: + title: Widget Service + version: 1.0.0 +paths: + /widgets/compose: + post: + operationId: ComposeWidget + requestBody: + content: + application/json: + schema: + $ref: '#/components/schemas/ComposeRequest' + responses: + "200": + description: ok + content: + application/json: + schema: + $ref: '#/components/schemas/Widget' + /widgets/compose/stream: + post: + operationId: StreamWidget + requestBody: + content: + application/json: + schema: + $ref: '#/components/schemas/ComposeRequest' + responses: + "200": + description: events + content: + text/event-stream: + schema: + type: object + required: [data] + properties: + data: + $ref: '#/components/schemas/WidgetEvent' + /widgets/{id}: + get: + operationId: GetWidget + parameters: + - name: id + in: path + required: true + schema: + type: string + responses: + "200": + description: ok + content: + application/json: + schema: + $ref: '#/components/schemas/Widget' +components: + schemas: + ComposeRequest: + type: object + properties: + prompt: + type: string + resumeToken: + type: string + previousId: + type: string + budget: + type: integer + options: + type: object + properties: + cursor: + type: string + required: [prompt] + WidgetStatus: + type: string + enum: [in_progress, more_required, budget_reached, completed, failed] + Widget: + type: object + properties: + id: + type: string + status: + $ref: '#/components/schemas/WidgetStatus' + resumeToken: + type: string + steps: + type: array + items: + type: object + properties: + text: + type: string + usage: + type: object + properties: + inputUnits: + type: integer + outputUnits: + type: integer + WidgetEvent: + type: object + properties: + event_type: + type: string + resumeToken: + type: string + delta: + type: object + properties: + text: + type: string + widget: + $ref: '#/components/schemas/Widget' +` + +const validContinuationManifest = ` +version: 1 +commands: + compose: + op: ComposeWidget + args: + prompt: {to: $.prompt, variadic: true} + flags: + resume-token: {to: $.resumeToken} + previous-id: {to: $.previousId} + watch: + op: StreamWidget + args: + prompt: {to: $.prompt, variadic: true} + output: + stream: + select: $.data.delta.text + outcome: + streamStatePointer: $.data.widget.status + states: {in_progress: pending, more_required: continue, budget_reached: success, completed: success, failed: failure} + continuation: + bind: + - streamFrom: [$.data.resumeToken, $.data.widget.resumeToken] + to: $.resumeToken + - streamFrom: $.data.widget.id + to: $.previousId + optional: true + maxHops: 4 + default: false +operations: + ComposeWidget: + output: + outcome: + statePointer: $.status + states: {in_progress: pending, more_required: continue, budget_reached: success, completed: success, failed: failure} + continuation: + bind: + - from: $.resumeToken + to: $.resumeToken + - from: $.id + to: $.previousId + optional: true + merge: $.steps + sum: [$.usage.inputUnits, $.usage.outputUnits] +` + +func TestCLICommands_ContinuationDecodeAndLink(t *testing.T) { + manifest, warnings, err := decodeCLIWithSpec(t, cliContinuationTestSpec, validContinuationManifest) + require.NoError(t, err) + assert.Empty(t, warnings) + require.Len(t, manifest.Operations, 1) + require.Len(t, manifest.Commands, 2) + + op := manifest.Operations[0].Continuation + require.NotNil(t, op) + assert.Equal(t, "ComposeWidget", op.OperationID) + assert.Equal(t, 32, op.MaxHops) + assert.True(t, op.Default) + assert.Equal(t, []string{"/steps"}, op.Merge) + assert.Equal(t, []string{"/usage/inputUnits", "/usage/outputUnits"}, op.Sum) + require.Len(t, op.Bind, 2) + assert.Equal(t, CLICommandContinuationBinding{From: "$.resumeToken", Pointer: "/resumeToken", To: "$.resumeToken", ToPointer: "/resumeToken", ToPath: []string{"resumeToken"}}, op.Bind[0]) + assert.True(t, op.Bind[1].Optional) + assert.Empty(t, op.Bind[1].Flag, "the operation entry declares no flags") + + compose := manifest.Commands[0].Continuation + require.NotNil(t, compose, "a single-route intent inherits its operation entry's recipe") + assert.Equal(t, "resume-token", compose.Bind[0].Flag) + assert.Equal(t, "previous-id", compose.Bind[1].Flag) + assert.NotSame(t, &op.Bind[0], &compose.Bind[0]) + + watch := manifest.Commands[1].Continuation + require.NotNil(t, watch) + assert.Equal(t, 4, watch.MaxHops) + assert.False(t, watch.Default) + assert.Equal(t, []string{"/data/resumeToken", "/data/widget/resumeToken"}, watch.Bind[0].StreamPointers) + assert.Equal(t, []string{"/data/widget/id"}, watch.Bind[1].StreamPointers) +} + +func TestCLICommands_ContinuationNestedTarget(t *testing.T) { + manifest, _, err := decodeCLIWithSpec(t, cliContinuationTestSpec, strings.Replace(validContinuationManifest, " - from: $.resumeToken\n to: $.resumeToken\n", " - from: $.resumeToken\n to: $.options.cursor\n", 1)) + require.NoError(t, err) + assert.Equal(t, []string{"options", "cursor"}, manifest.Operations[0].Continuation.Bind[0].ToPath) + assert.Equal(t, "/options/cursor", manifest.Operations[0].Continuation.Bind[0].ToPointer) +} + +func TestCLICommands_ContinuationErrors(t *testing.T) { + base := validContinuationManifest + opOutcome := " output:\n outcome:\n statePointer: $.status\n states: {in_progress: pending, more_required: continue, budget_reached: success, completed: success, failed: failure}\n" + for _, tc := range []struct { + name string + manifest string + want string + }{ + {"unknown key", strings.Replace(base, " maxHops: 4\n", " maxHops: 4\n hops: 2\n", 1), `continuation has unknown key "hops"`}, + {"unknown bind key", strings.Replace(base, " to: $.previousId\n optional: true\n maxHops", " to: $.previousId\n required: false\n maxHops", 1), `continuation.bind has unknown key "required"`}, + {"missing bind", strings.Replace(base, " bind:\n - from: $.resumeToken\n to: $.resumeToken\n - from: $.id\n to: $.previousId\n optional: true\n", "", 1), "continuation requires bind"}, + {"only optional", strings.Replace(base, " - from: $.resumeToken\n to: $.resumeToken\n", " - from: $.resumeToken\n to: $.resumeToken\n optional: true\n", 1), "at least one binding that is not optional"}, + {"duplicate target", strings.Replace(base, " to: $.previousId\n optional: true\n merge", " to: $.resumeToken\n optional: true\n merge", 1), "writes $.resumeToken more than once"}, + {"missing to", strings.Replace(base, " - from: $.resumeToken\n to: $.resumeToken\n", " - from: $.resumeToken\n", 1), "requires to (the request-body path"}, + {"missing from", strings.Replace(base, " - from: $.resumeToken\n to: $.resumeToken\n", " - to: $.resumeToken\n", 1), "requires from (JSON response) or streamFrom"}, + {"indexed to", strings.Replace(base, " to: $.resumeToken\n - from: $.id", " to: $.options[0]\n - from: $.id", 1), "must name object properties only"}, + {"zero hops", strings.Replace(base, "maxHops: 4", "maxHops: 0", 1), "continuation.maxHops must be at least 1"}, + {"hops not an int", strings.Replace(base, "maxHops: 4", "maxHops: many", 1), "continuation.maxHops must be an integer"}, + {"no outcome", strings.Replace(base, opOutcome, "", 1), "declares continuation but no output.outcome"}, + {"no continue state", strings.Replace(base, "states: {in_progress: pending, more_required: continue, budget_reached: success, completed: success, failed: failure}\n continuation", "states: {in_progress: pending, more_required: success, budget_reached: success, completed: success, failed: failure}\n continuation", 1), "classify at least one state as continue"}, + {"from not a string", strings.Replace(base, " - from: $.resumeToken\n", " - from: $.steps\n", 1), "selected value must be a string"}, + {"from on stream op", strings.Replace(base, " - streamFrom: [$.data.resumeToken, $.data.widget.resumeToken]\n", " - from: $.resumeToken\n", 1), "has no application/json 2xx response; use streamFrom"}, + {"streamFrom misses", strings.Replace(base, "streamFrom: $.data.widget.id", "streamFrom: $.data.widget.ident", 1), "does not resolve in any event shape"}, + {"to absent", strings.Replace(base, " - from: $.resumeToken\n to: $.resumeToken\n", " - from: $.resumeToken\n to: $.resumeTokn\n", 1), `property "resumeTokn" is absent from request body variant`}, + {"to not a string", strings.Replace(base, " - from: $.resumeToken\n to: $.resumeToken\n", " - from: $.resumeToken\n to: $.budget\n", 1), "must name a string property"}, + {"merge not an array", strings.Replace(base, "merge: $.steps", "merge: $.id", 1), "continuation.merge $.id in operation \"ComposeWidget\" response 200 must select an array"}, + {"sum not a number", strings.Replace(base, "sum: [$.usage.inputUnits, $.usage.outputUnits]", "sum: [$.id]", 1), "must select a number"}, + {"indexed merge", strings.Replace(base, "merge: $.steps", "merge: $.steps[0].parts", 1), "continuation.merge $.steps[0].parts must name object properties only"}, + {"indexed sum", strings.Replace(base, "sum: [$.usage.inputUnits, $.usage.outputUnits]", "sum: [$.usage.inputUnits, \"$.steps[0].units\"]", 1), "continuation.sum $.steps[0].units must name object properties only"}, + {"merge on stream op", strings.Replace(base, " maxHops: 4\n", " maxHops: 4\n merge: $.data.widget.steps\n", 1), "apply to application/json responses"}, + {"bad classification", strings.Replace(base, "more_required: continue, budget_reached: success, completed: success, failed: failure}\n continuation", "more_required: resume, budget_reached: success, completed: success, failed: failure}\n continuation", 1), `has unsupported classification "resume"`}, + {"reserved flag", strings.Replace(base, " previous-id: {to: $.previousId}\n", " previous-id: {to: $.previousId}\n auto-continue: {to: $.prompt}\n", 1), "auto-continue"}, + {"async conflict", strings.Replace(base, " watch:\n", " start:\n op: ComposeWidget\n async:\n op: GetWidget\n id: {from: $.id, to: {in: path, name: id}}\n statePointer: $.status\n states: {in_progress: pending, more_required: pending, budget_reached: success, completed: success, failed: failure}\n interval: 1s\n maxInterval: 1s\n timeout: 1m\n watch:\n", 1), `async command "start" also uses as its create operation`}, + {"async and continuation", strings.Replace(base, " watch:\n op: StreamWidget\n", " watch:\n op: StreamWidget\n async:\n op: GetWidget\n id: {from: $.id, to: {in: path, name: id}}\n statePointer: $.status\n states: {completed: success}\n", 1), "declares both async and output.stream"}, + {"planned command", "version: 1\ncommands:\n later:\n planned: later\n continuation:\n bind: [{from: $.id, to: $.previousId}]\n", "planned command and cannot declare continuation"}, + {"command continue state without recipe", strings.Replace(base, " continuation:\n bind:\n - streamFrom: [$.data.resumeToken, $.data.widget.resumeToken]\n to: $.resumeToken\n - streamFrom: $.data.widget.id\n to: $.previousId\n optional: true\n maxHops: 4\n default: false\n", "", 1), `command "watch" classifies output.outcome state "more_required" as continue but declares no continuation recipe`}, + {"operation continue state without recipe", "version: 1\noperations:\n GetWidget:\n output:\n outcome:\n statePointer: $.status\n states: {in_progress: pending, more_required: continue, budget_reached: success, completed: success, failed: failure}\n", `operation "GetWidget" classifies output.outcome state "more_required" as continue but declares no continuation recipe`}, + {"operation continue state with recipe only on the intent", "version: 1\ncommands:\n compose:\n op: ComposeWidget\n args:\n prompt: {to: $.prompt, variadic: true}\n flags:\n resume-token: {to: $.resumeToken}\n continuation:\n bind:\n - from: $.resumeToken\n to: $.resumeToken\noperations:\n ComposeWidget:\n" + opOutcome, `operation "ComposeWidget" classifies output.outcome state "more_required" as continue but declares no continuation recipe; an operation entry's outcome also applies to the generated operation command`}, + } { + t.Run(tc.name, func(t *testing.T) { + _, _, err := decodeCLIWithSpec(t, cliContinuationTestSpec, tc.manifest) + require.Error(t, err) + assert.Contains(t, err.Error(), tc.want) + }) + } +} + +func TestCLICommands_ContinuationTargetSharedByUnionMembers(t *testing.T) { + spec := strings.Replace(cliContinuationTestSpec, ` ComposeRequest: + type: object + properties: + prompt: + type: string + resumeToken: + type: string + previousId: + type: string + budget: + type: integer + options: + type: object + properties: + cursor: + type: string + required: [prompt] +`, ` ComposeRequest: + type: object + properties: + resumeToken: + type: string + oneOf: + - $ref: '#/components/schemas/ComposeText' + - $ref: '#/components/schemas/ComposeImage' + ComposeText: + type: object + properties: + prompt: + type: string + ComposeImage: + type: object + properties: + image: + type: string +`, 1) + require.NotEqual(t, cliContinuationTestSpec, spec) + manifest, _, err := decodeCLIWithSpec(t, spec, ` +version: 1 +operations: + ComposeWidget: + output: + outcome: + statePointer: $.status + states: {in_progress: pending, more_required: continue, budget_reached: success, completed: success, failed: failure} + continuation: + bind: + - from: $.resumeToken + to: $.resumeToken +`) + require.NoError(t, err) + assert.Equal(t, []string{"resumeToken"}, manifest.Operations[0].Continuation.Bind[0].ToPath) +} + +func TestCLICommands_ContinuationTargetSchemas(t *testing.T) { + manifest := ` +version: 1 +operations: + ComposeWidget: + output: + outcome: + statePointer: $.status + states: {in_progress: pending, more_required: continue, budget_reached: success, completed: success, failed: failure} + continuation: + bind: + - from: $.resumeToken + to: $.options.cursor +` + for _, tc := range []struct { + name string + options string + wantErr string + }{ + {"nullable type list", ` + options: + type: object + properties: + cursor: + type: [string, "null"] +`, ""}, + {"nullable anyOf", ` + options: + type: object + properties: + cursor: + anyOf: + - type: string + - type: "null" +`, ""}, + {"readOnly arm of a nullable union", ` + options: + type: object + properties: + cursor: + anyOf: + - type: string + readOnly: true + - type: "null" +`, `names readOnly property "cursor"`}, + {"string or integer union", ` + options: + type: object + properties: + cursor: + anyOf: + - type: string + - type: integer +`, "must name a string property"}, + {"nested object union", ` + options: + oneOf: + - type: object + properties: + cursor: {type: string} + mode: {type: string, const: fast} + - type: object + properties: + cursor: {type: string} + depth: {type: integer} + - type: "null" +`, ""}, + {"nested union member without the target", ` + options: + oneOf: + - type: object + properties: + cursor: {type: string} + - type: object + properties: + depth: {type: integer} +`, `property "cursor" is absent from request body variant`}, + {"nested union member with a non-string target", ` + options: + oneOf: + - type: object + properties: + cursor: {type: string} + - type: object + properties: + cursor: {type: integer} +`, "must name a string property"}, + } { + t.Run(tc.name, func(t *testing.T) { + spec := strings.Replace(cliContinuationTestSpec, ` + options: + type: object + properties: + cursor: + type: string +`, tc.options, 1) + require.NotEqual(t, cliContinuationTestSpec, spec) + decoded, _, err := decodeCLIWithSpec(t, spec, manifest) + if tc.wantErr != "" { + require.Error(t, err) + assert.Contains(t, err.Error(), tc.wantErr) + return + } + require.NoError(t, err) + assert.Equal(t, []string{"options", "cursor"}, decoded.Operations[0].Continuation.Bind[0].ToPath) + }) + } +} + +func TestCLICommands_ContinuationAsyncStatesRejectContinue(t *testing.T) { + _, _, err := decodeCLIWithSpec(t, cliContinuationTestSpec, ` +version: 1 +commands: + start: + op: ComposeWidget + async: + op: GetWidget + id: {from: $.id, to: {in: path, name: id}} + statePointer: $.status + states: {in_progress: pending, more_required: continue, budget_reached: success, completed: success, failed: failure} + interval: 1s + maxInterval: 1s + timeout: 1m +`) + require.Error(t, err) + assert.Contains(t, err.Error(), `async.states["more_required"] has unsupported classification "continue"`) +} + func TestCLICommands_FlagImplies(t *testing.T) { manifest, _, err := decodeCLITest(t, ` version: 1 diff --git a/templates/perms.go b/templates/perms.go index ca046ab3a..f7663cb1c 100644 --- a/templates/perms.go +++ b/templates/perms.go @@ -55,6 +55,7 @@ var fileMode = map[string]fs.FileMode{ "templates/cli/auxiliary/internal/output/binary.go.stmpl": 0o644, "templates/cli/auxiliary/internal/output/classify.go.stmpl": 0o644, "templates/cli/auxiliary/internal/output/color.go.stmpl": 0o644, + "templates/cli/auxiliary/internal/output/continuation.go.stmpl": 0o644, "templates/cli/auxiliary/internal/output/defaulttimeout.go.stmpl": 0o644, "templates/cli/auxiliary/internal/output/errortable.go.stmpl": 0o644, "templates/cli/auxiliary/internal/output/exitcodes.go.stmpl": 0o644, @@ -182,6 +183,7 @@ var fileMode = map[string]fs.FileMode{ "templates/cli/tests/primary/collections_additional_test.go.stmpl": 0o644, "templates/cli/tests/primary/compact_help_test.go.stmpl": 0o644, "templates/cli/tests/primary/configure_test.go.stmpl": 0o644, + "templates/cli/tests/primary/continuation_additional_test.go.stmpl": 0o644, "templates/cli/tests/primary/custom_commands_test.go.stmpl": 0o644, "templates/cli/tests/primary/diagnostics_test.go.stmpl": 0o644, "templates/cli/tests/primary/dispatch_test.go.stmpl": 0o644, diff --git a/templates/templates/cli/README.md b/templates/templates/cli/README.md index 8948d0f91..f1f05affe 100644 --- a/templates/templates/cli/README.md +++ b/templates/templates/cli/README.md @@ -870,7 +870,7 @@ Dry-run does not depend on agent environment detection. Successful previews use **CLI-originated errors** (`ExecuteRoot` in `root.go.stmpl`, `CLIError` in `agentmode.go.stmpl`): every CLI error is classified, including compatibility-mode errors that main still prints. Flag/argument/JSON parsing failures, Cobra usage errors, invalid option values, and local input/config failures are typed as `CLI_VALIDATION` at their structural boundaries (`flagutil.WithCLIValidation`); an error carrying a `clierrors.ExitCoder` classifies from its code (`ExitAuth` → `CLI_AUTHENTICATION`, `ExitUsage` → `CLI_VALIDATION`). `AgentModeError(cmd, message, hints)` is `validation_error` / `CLI_VALIDATION`. Runtimes may attach `CLIReason()` and `CLIHints()`; command hints append after them. `output.InstallErrorHandling` composes final-tree Cobra flag/Args handlers and pre-validates required/flag-group contracts; the root gets `cobra.NoArgs` so cobra's root-only `legacyArgs` check never runs, and the `Args` wrapper turns a rejected positional on any command with subcommands into `output.UnknownSubcommandError` (one-line message, suggestions as hints). On parent commands that reject positionals (the root, generated and synthesized groups), the flag error handler validates the positionals it already has through the same `Args` path first, so a typo followed by the intended leaf's flags reports the command rather than the flag; leaves and parents that accept positionals keep the flag error. Because the root no longer relies on `root.Find` rejecting stray tokens, `installHelpTopicCheck` wraps the built-in `help` command so `cli help genreate` still prints `Unknown help topic` plus root usage (exit 0, as cobra did). `output.UsageHelpError` is the reusable exported helper for incomplete hand-written/generated commands: plain mode writes full help to stderr then `Error: ...`; agent mode, explicit JSON or `--jq`, and explicit TOON when `classifiedErrors` is enabled return one envelope with a command-specific `--help` hint. `ExecuteRoot` snapshots rendering flags before Cobra can reject an early command. Rendered diagnostics return one sentinel containing the classification, so `main.go` never prints a second copy. -The closed reason namespace is `CLI_VALIDATION`, `CLI_CONNECTION`, `CLI_PROTOCOL`, `CLI_RUNTIME`, `CLI_UNAVAILABLE`, and `CLI_AUTHENTICATION`; it mirrors `CLIRuntimeHintReasons` in `internal/extensions/cli_commands.go`. Interactive prompt cancellation and TUI/explorer failures deliberately remain untyped and therefore exit as runtime failures. +The closed reason namespace is `CLI_VALIDATION`, `CLI_CONNECTION`, `CLI_PROTOCOL`, `CLI_RUNTIME`, `CLI_UNAVAILABLE`, `CLI_AUTHENTICATION`, `CLI_ASYNC_FAILED`, `CLI_ASYNC_TIMEOUT`, `CLI_ASYNC_UNKNOWN_STATE`, `CLI_OUTCOME_FAILED`, and `CLI_CONTINUATION_LIMIT`; it mirrors `CLIRuntimeHintReasons` in `internal/extensions/cli_commands.go`. Interactive prompt cancellation and TUI/explorer failures deliberately remain untyped and therefore exit as runtime failures. ### Exit Codes and Error Boundaries diff --git a/templates/templates/cli/auxiliary/internal/output/classify.go.stmpl b/templates/templates/cli/auxiliary/internal/output/classify.go.stmpl index 08b684a58..fc531221e 100644 --- a/templates/templates/cli/auxiliary/internal/output/classify.go.stmpl +++ b/templates/templates/cli/auxiliary/internal/output/classify.go.stmpl @@ -703,6 +703,7 @@ const ( ReasonCLIAsyncTimeout = "CLI_ASYNC_TIMEOUT" ReasonCLIAsyncUnknownState = "CLI_ASYNC_UNKNOWN_STATE" ReasonCLIOutcomeFailed = "CLI_OUTCOME_FAILED" + ReasonCLIContinuationLimit = "CLI_CONTINUATION_LIMIT" ) type cliReasonError struct { @@ -725,7 +726,7 @@ func normalizeCLIReason(reason string) string { case ReasonCLIValidation, ReasonCLIConnection, ReasonCLIProtocol, ReasonCLIRuntime, ReasonCLIUnavailable, ReasonCLIAuthentication, ReasonCLIAsyncFailed, ReasonCLIAsyncTimeout, ReasonCLIAsyncUnknownState, - ReasonCLIOutcomeFailed: + ReasonCLIOutcomeFailed, ReasonCLIContinuationLimit: return reason } return ReasonCLIRuntime diff --git a/templates/templates/cli/auxiliary/internal/output/continuation.go.stmpl b/templates/templates/cli/auxiliary/internal/output/continuation.go.stmpl new file mode 100644 index 000000000..5be2891fa --- /dev/null +++ b/templates/templates/cli/auxiliary/internal/output/continuation.go.stmpl @@ -0,0 +1,617 @@ +{{- if hasCliContinuations}} +package output + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "io" + "math/big" + "net/http" + "reflect" + "sort" + "strconv" + "strings" + + "github.com/spf13/cobra" + + "{{getGolangPackage}}/internal/flagutil" +) + +const continuationAnnotation = "speakeasy_continuation" + +type continuationBinding struct { + Pointer string `json:"pointer"` + StreamPointers []string `json:"streamPointers"` + To string `json:"to"` + ToPath []string `json:"toPath"` + Optional bool `json:"optional"` + Flag string `json:"flag"` +} + +type continuationConfig struct { + Bind []continuationBinding `json:"bind"` + Merge []string `json:"merge"` + Sum []string `json:"sum"` + MaxHops int `json:"maxHops"` +} + +// ContinuationHop sends the command's request once more. The request the +// driver mutates between hops is the one the closure sends. +type ContinuationHop func() (interface{}, error) + +func activeContinuation(cmd *cobra.Command) *continuationConfig { + if cmd == nil || cmd.Annotations == nil { + return nil + } + raw := cmd.Annotations[continuationAnnotation] + if raw == "" { + return nil + } + var cfg continuationConfig + if err := json.Unmarshal([]byte(raw), &cfg); err != nil || len(cfg.Bind) == 0 { + return nil + } + return &cfg +} + +// ContinuationActive reports whether the command declares a continuation +// recipe the runtime can drive. +func ContinuationActive(cmd *cobra.Command) bool { + return activeContinuation(cmd) != nil && activeOutcome(cmd) != nil +} + +type continuationRun struct { + cmd *cobra.Command + cfg *continuationConfig + outcome *outcomeConfig + request interface{} + hop ContinuationHop + auto bool + limit int + + hops int + state string + values map[string]string + limited bool + missing []string +} + +func newContinuationRun(cmd *cobra.Command, request interface{}, hop ContinuationHop) (*continuationRun, error) { + run := &continuationRun{cmd: cmd, cfg: activeContinuation(cmd), outcome: activeOutcome(cmd), request: request, hop: hop, values: map[string]string{}} + run.auto, _ = flagutil.GetBoolFlag(cmd, "auto-continue") + run.limit = run.cfg.MaxHops + if flagutil.FlagChanged(cmd, "max-continuations") { + limit, _ := flagutil.GetIntFlag(cmd, "max-continuations") + if limit < 1 { + return nil, flagutil.WithCLIValidation(fmt.Errorf("--max-continuations must be at least 1, got %d", limit)) + } + run.limit = limit + } + if run.limit < 1 { + run.limit = 1 + } + return run, nil +} + +func (r *continuationRun) beginHop() { + r.state = "" + r.values = map[string]string{} +} + +func (r *continuationRun) capture(payload interface{}, stream bool) { + for _, binding := range r.cfg.Bind { + pointers := []string{binding.Pointer} + if stream { + pointers = binding.StreamPointers + } + for _, pointer := range pointers { + if pointer == "" { + continue + } + if value, ok := outcomePointer(payload, pointer); ok { + if text, isString := value.(string); isString && text != "" { + r.values[binding.To] = text + } + } + } + } + pointer := r.outcome.StatePointer + if stream { + pointer = r.outcome.StreamStatePointer + } + if value, ok := outcomePointer(payload, pointer); ok { + if text, isString := value.(string); isString && text != "" { + r.state = text + } + } +} + +// advance reports whether another hop should be sent and, when it should, +// writes the resume values into the request. +func (r *continuationRun) advance() (bool, error) { + if r.outcome.States[r.state] != "continue" { + return false, nil + } + r.missing = nil + for _, binding := range r.cfg.Bind { + if !binding.Optional && r.values[binding.To] == "" { + r.missing = append(r.missing, binding.To) + } + } + if len(r.missing) > 0 || !r.auto { + return false, nil + } + if r.hops >= r.limit { + r.limited = true + return false, nil + } + for _, binding := range r.cfg.Bind { + value, ok := r.values[binding.To] + if !ok { + continue + } + if err := setContinuationBodyValue(r.request, binding.ToPath, value); err != nil { + return false, r.wrap(WithCLIReason(fmt.Errorf("cannot set %s on the next request: %w", binding.To, err), ReasonCLIRuntime)) + } + } + return true, nil +} + +func (r *continuationRun) send() (interface{}, error) { + res, err := r.hop() + if err != nil { + if r.hops == 0 { + return nil, err + } + return nil, r.wrap(err) + } + r.hops++ + return res, nil +} + +func (r *continuationRun) resumeFields() map[string]interface{} { + if !r.resumable() { + return nil + } + fields := map[string]interface{}{} + for to, value := range r.values { + fields[to] = value + } + return fields +} + +func (r *continuationRun) resumable() bool { + for _, binding := range r.cfg.Bind { + if !binding.Optional && r.values[binding.To] != "" { + return true + } + } + return false +} + +// resumeHint names the command's own flags for each resume value; values +// without a declared flag are named by their request-body path. +func (r *continuationRun) resumeHint() string { + if !r.resumable() { + return "" + } + var flags, paths []string + for _, binding := range r.cfg.Bind { + value, ok := r.values[binding.To] + if !ok { + continue + } + if binding.Flag != "" { + flags = append(flags, fmt.Sprintf("--%s %s", binding.Flag, strconv.Quote(value))) + } else { + paths = append(paths, fmt.Sprintf("%s=%s", binding.To, strconv.Quote(value))) + } + } + var parts []string + if len(flags) > 0 { + parts = append(parts, "re-run the same command with "+strings.Join(flags, " ")) + } + if len(paths) > 0 { + parts = append(parts, "set "+strings.Join(paths, ", ")+" in the request body") + } + return "To continue, " + strings.Join(parts, " and ") +} + +func (r *continuationRun) wrap(err error) error { + if err == nil { + return nil + } + return &continuationError{err: err, hops: r.hops, resume: r.resumeFields(), hint: r.resumeHint()} +} + +// finish reports what the loop stopped on once the result is rendered. +func (r *continuationRun) finish() error { + if r.limited { + return Error(r.cmd, &continuationLimitError{hops: r.hops, resume: r.resumeFields(), hint: r.resumeHint()}) + } + if len(r.missing) > 0 { + sort.Strings(r.missing) + return Error(r.cmd, r.wrap(WithCLIReason(fmt.Errorf("the response reported state %q but carried no %s to continue with; the output is incomplete", r.state, strings.Join(r.missing, ", ")), ReasonCLIProtocol))) + } + if IsMachineMode(r.cmd) || r.cmd.Context().Err() != nil { + return nil + } + stderr := r.cmd.ErrOrStderr() + if r.hops > 1 { + fmt.Fprintf(stderr, "Continued across %d responses.\n", r.hops) + } + if hint := r.resumeHint(); hint != "" { + fmt.Fprintln(stderr, hint+".") + } + return nil +} + +type continuationError struct { + err error + hops int + resume map[string]interface{} + hint string +} + +func (e *continuationError) Error() string { return e.err.Error() } + +func (e *continuationError) Unwrap() error { return e.err } + +func (e *continuationError) MachineErrorFields() map[string]interface{} { + fields := map[string]interface{}{} + var inner interface{ MachineErrorFields() map[string]interface{} } + if errors.As(e.err, &inner) { + for key, value := range inner.MachineErrorFields() { + fields[key] = value + } + } + fields["hops"] = e.hops + if len(e.resume) > 0 { + fields["continuation"] = e.resume + } + return fields +} + +func (e *continuationError) CLIHints() []string { + var hints []string + var inner interface{ CLIHints() []string } + if errors.As(e.err, &inner) { + hints = append(hints, inner.CLIHints()...) + } + if e.hint != "" { + hints = append(hints, e.hint) + } + return hints +} + +type continuationLimitError struct { + hops int + resume map[string]interface{} + hint string +} + +func (e *continuationLimitError) Error() string { + return fmt.Sprintf("stopped after %d responses; the response still asks to continue (raise --max-continuations to allow more)", e.hops) +} + +func (e *continuationLimitError) CLIReason() string { return ReasonCLIContinuationLimit } + +func (e *continuationLimitError) MachineErrorFields() map[string]interface{} { + fields := map[string]interface{}{"hops": e.hops} + if len(e.resume) > 0 { + fields["continuation"] = e.resume + } + return fields +} + +func (e *continuationLimitError) CLIHints() []string { + if e.hint == "" { + return nil + } + return []string{e.hint} +} + +// ContinuationResult sends the request, then re-sends it with the resume +// values from each response while the response's state classifies as +// continue, and renders one result. +func ContinuationResult(cmd *cobra.Command, request interface{}, hop ContinuationHop) error { + run, err := newContinuationRun(cmd, request, hop) + if err != nil { + return err + } + res, err := run.send() + if err != nil { + return Error(cmd, err) + } + return run.discrete(res) +} + +// discrete continues from an already-sent response whose body is a single +// JSON document. +func (r *continuationRun) discrete(res interface{}) error { + cmd, run := r.cmd, r + var ( + last interface{} + payloads []interface{} + ) + for { + if len(payloads) > 0 { + var err error + if res, err = run.send(); err != nil { + return Error(cmd, err) + } + } + if flagutil.DidDryRunRequest(cmd) { + return Result(cmd, res) + } + payload, ok := outcomeResultPayload(res) + if !ok { + return Result(cmd, res) + } + last = res + payloads = append(payloads, payload) + run.beginHop() + run.capture(payload, false) + more, err := run.advance() + if err != nil { + return Error(cmd, err) + } + if !more { + break + } + } + + target := last + if len(payloads) > 1 && (len(run.cfg.Merge) > 0 || len(run.cfg.Sum) > 0) { + target = mergedContinuationResult(last, mergeContinuationPayloads(run.cfg, payloads)) + } + if failure := outcomeResultFailure(cmd, target); failure != nil { + {{- if hasCliArtifacts}} + if activeArtifact(cmd) != nil { + return Error(cmd, run.wrap(failure)) + } + {{- end}} + if err := renderResult(cmd, target, true); err != nil { + return err + } + return Error(cmd, run.wrap(failure)) + } + if err := Result(cmd, target); err != nil { + return err + } + return run.finish() +} + +func mergeContinuationPayloads(cfg *continuationConfig, payloads []interface{}) interface{} { + merged := payloads[len(payloads)-1] + for _, pointer := range cfg.Merge { + var items []interface{} + found := false + for _, payload := range payloads { + if value, ok := outcomePointer(payload, pointer); ok { + if list, isList := value.([]interface{}); isList { + items = append(items, list...) + found = true + } + } + } + if found { + merged = setContinuationPointer(merged, pointer, items) + } + } + for _, pointer := range cfg.Sum { + if total, ok := sumContinuationValues(payloads, pointer); ok { + merged = setContinuationPointer(merged, pointer, total) + } + } + return merged +} + +func sumContinuationValues(payloads []interface{}, pointer string) (json.Number, bool) { + total := new(big.Rat) + found := false + for _, payload := range payloads { + value, ok := outcomePointer(payload, pointer) + if !ok { + continue + } + number, isNumber := value.(json.Number) + if !isNumber { + continue + } + term, ok := new(big.Rat).SetString(number.String()) + if !ok { + continue + } + total.Add(total, term) + found = true + } + if !found { + return "", false + } + if total.IsInt() { + return json.Number(total.Num().String()), true + } + approx, _ := total.Float64() + return json.Number(strconv.FormatFloat(approx, 'g', -1, 64)), true +} + +func setContinuationPointer(root interface{}, pointer string, value interface{}) interface{} { + if !strings.HasPrefix(pointer, "/") { + return root + } + var tokens []string + for _, raw := range strings.Split(strings.TrimPrefix(pointer, "/"), "/") { + tokens = append(tokens, strings.ReplaceAll(strings.ReplaceAll(raw, "~1", "/"), "~0", "~")) + } + object, ok := root.(map[string]interface{}) + if !ok { + return root + } + current := object + for _, token := range tokens[:len(tokens)-1] { + child, ok := current[token].(map[string]interface{}) + if !ok { + child = map[string]interface{}{} + current[token] = child + } + current = child + } + current[tokens[len(tokens)-1]] = value + return object +} + +type continuationHTTPMeta struct { + Response *http.Response +} + +type continuationResultEnvelope struct { + HTTPMeta continuationHTTPMeta + Result interface{} +} + +func mergedContinuationResult(last interface{}, merged interface{}) interface{} { + raw, err := json.Marshal(merged) + if err != nil { + return last + } + return &continuationResultEnvelope{ + HTTPMeta: continuationHTTPMeta{Response: &http.Response{ + Body: io.NopCloser(bytes.NewReader(raw)), + Header: extractResponseHeaders(last).Clone(), + }}, + Result: merged, + } +} + +// setContinuationBodyValue writes value at path inside the request body: the +// request field tagged as the media-type body, or the request itself when the +// body is the whole request. The body round-trips through JSON so nested +// paths and union bodies are written the way the API receives them. +func setContinuationBodyValue(request interface{}, path []string, value string) error { + target, err := continuationBodyTarget(request) + if err != nil { + return err + } + if target.Kind() == reflect.Ptr && target.IsNil() { + target.Set(reflect.New(target.Type().Elem())) + } + encoded, err := json.Marshal(target.Interface()) + if err != nil { + return err + } + decoded, err := decodeWithNumbers(encoded) + if err != nil { + return err + } + body, ok := decoded.(map[string]interface{}) + if !ok { + return errors.New("the request body is not a JSON object") + } + flagutil.SetBodyPath(body, path, value) + encoded, err = json.Marshal(body) + if err != nil { + return err + } + fresh := reflect.New(target.Type()) + if err := json.Unmarshal(encoded, fresh.Interface()); err != nil { + return err + } + target.Set(fresh.Elem()) + return nil +} + +func continuationBodyTarget(request interface{}) (reflect.Value, error) { + v := reflect.ValueOf(request) + if v.Kind() != reflect.Ptr || v.IsNil() { + return reflect.Value{}, errors.New("the request must be passed by pointer") + } + v = v.Elem() + s := v + for s.Kind() == reflect.Ptr && !s.IsNil() { + s = s.Elem() + } + if s.Kind() == reflect.Struct { + for i := 0; i < s.NumField(); i++ { + if strings.HasPrefix(s.Type().Field(i).Tag.Get("request"), "mediaType=") && s.Field(i).CanSet() { + return s.Field(i), nil + } + } + } + return v, nil +} + +{{- if anyOperationHasStreaming}} + +// ContinuationStream streams the response, then opens the next hop's stream +// with the resume values seen in the events while the last reported state +// classifies as continue. Every hop's events go to one output sequence. +func ContinuationStream(cmd *cobra.Command, request interface{}, hop ContinuationHop, streamFieldName string) error { + run, err := newContinuationRun(cmd, request, hop) + if err != nil { + return err + } + if run.outcome.StreamStatePointer == "" { + res, err := run.send() + if err != nil { + return Error(cmd, err) + } + return StreamResult(cmd, res, streamFieldName) + } + res, err := run.send() + if err != nil { + return Error(cmd, err) + } + run.beginHop() + hops := &streamContinuation{run: run} + if err := streamResult(cmd, res, streamFieldName, hops, nil); err != nil || hops.finished { + return err + } + return run.finish() +} + +type streamContinuation struct { + run *continuationRun + finished bool +} + +func (s *streamContinuation) discrete(res interface{}) error { + s.finished = true + return s.run.discrete(res) +} + +func (s *streamContinuation) observe(frame map[string]interface{}) { + if frame != nil { + s.run.capture(frame, true) + } +} + +func (s *streamContinuation) wrap(err error) error { return s.run.wrap(err) } + +func (s *streamContinuation) next() (interface{}, bool, error) { + switch s.run.outcome.States[s.run.state] { + case "continue", "success", "failure": + default: + if s.run.resumable() { + message := "the stream ended before reporting a final state" + if s.run.state != "" { + message = fmt.Sprintf("the stream ended while the response was still %q", s.run.state) + } + return nil, false, s.run.wrap(WithCLIReason(errors.New(message), ReasonCLIConnection)) + } + } + more, err := s.run.advance() + if err != nil || !more { + return nil, false, err + } + res, err := s.run.send() + if err != nil { + return nil, false, err + } + s.run.beginHop() + return res, true, nil +} +{{- end}} +{{- end}} diff --git a/templates/templates/cli/auxiliary/internal/output/streaming.go.stmpl b/templates/templates/cli/auxiliary/internal/output/streaming.go.stmpl index 1f0384619..389c8a5cf 100644 --- a/templates/templates/cli/auxiliary/internal/output/streaming.go.stmpl +++ b/templates/templates/cli/auxiliary/internal/output/streaming.go.stmpl @@ -31,6 +31,22 @@ func StreamResult(cmd *cobra.Command, res interface{}, streamFieldName string, s } } } + return streamResult(cmd, res, streamFieldName, nil, stop) +} + +// streamHops continues a stream into the next hop's stream when one ends, +// so every hop's events reach one projector and one output sequence. +type streamHops interface { + observe(frame map[string]interface{}) + wrap(err error) error + next() (interface{}, bool, error) + discrete(res interface{}) error +} + +func streamResult(cmd *cobra.Command, res interface{}, streamFieldName string, hops streamHops, stop func()) error { + if stop == nil { + stop = func() {} + } defer stop() if flagutil.DidDryRunRequest(cmd) { return nil @@ -43,48 +59,6 @@ func StreamResult(cmd *cobra.Command, res interface{}, streamFieldName string, s jqExpr, _ := flagutil.GetStringFlag(cmd, "jq") jqRaw := jqRawOutput(cmd) - // Extract the stream field from the response envelope via reflection - v := reflect.ValueOf(res) - if v.Kind() == reflect.Ptr { - if v.IsNil() { - return nil - } - v = v.Elem() - } - if v.Kind() != reflect.Struct { - // Fallback: not a struct envelope — output as regular result - stop() - return Result(cmd, res) - } - streamVal := v.FieldByName(streamFieldName) - if !streamVal.IsValid() { - // Fallback: stream field not found — output as regular result - stop() - return Result(cmd, res) - } - // Guard IsNil: only call on nillable kinds. Fall back to Result() for - // mixed-content operations where non-stream content may still exist. - if isNillable(streamVal) && streamVal.IsNil() { - stop() - return Result(cmd, res) - } - - // Validate streaming interface: must have Next and Value methods - nextMethod := streamVal.MethodByName("Next") - 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 - stop() - return Result(cmd, res) - } - - if closeMethod.IsValid() { - defer closeMethod.Call(nil) - } - projector := newStreamProjector(cmd) outcomeCfg := activeOutcome(cmd) if outcomeCfg != nil && outcomeCfg.StreamStatePointer == "" { @@ -112,84 +86,166 @@ func StreamResult(cmd *cobra.Command, res interface{}, streamFieldName string, s } return rendered } + wrap := func(e error) error { + if hops != nil { + return hops.wrap(e) + } + return e + } first := true - for { - // Check for context cancellation (Ctrl+C) - select { - case <-ctx.Done(): - return finishJoin(ctx.Err()) - default: - } - - // Call Next() bool - nextResults := nextMethod.Call(nil) - stop() - if !nextResults[0].Bool() { - break + // drain writes one response's events. fallback reports a response that + // carries no stream, which is rendered as a regular result instead. + drain := func(res interface{}) (fallback bool, err error) { + // Extract the stream field from the response envelope via reflection + v := reflect.ValueOf(res) + if v.Kind() == reflect.Ptr { + if v.IsNil() { + return false, nil + } + v = v.Elem() + } + if v.Kind() != reflect.Struct { + return true, nil + } + streamVal := v.FieldByName(streamFieldName) + if !streamVal.IsValid() { + return true, nil + } + // Guard IsNil: only call on nillable kinds. Fall back to Result() for + // mixed-content operations where non-stream content may still exist. + if isNillable(streamVal) && streamVal.IsNil() { + return true, nil + } + + // Validate streaming interface: must have Next and Value methods + nextMethod := streamVal.MethodByName("Next") + valueMethod := streamVal.MethodByName("Value") + errMethod := streamVal.MethodByName("Err") + closeMethod := streamVal.MethodByName("Close") + rawMethod := streamVal.MethodByName("RawValue") + if !nextMethod.IsValid() || !valueMethod.IsValid() { + return true, nil } - // Call Value() — handles both: - // EventStream: Value() *T (1 return value) - // JsonLStream: Value() (T, error) (2 return values) - valueResults := valueMethod.Call(nil) - var item interface{} - if len(valueResults) == 2 { - // JsonLStream: check error — route through Error() for JSON formatting - if !valueResults[1].IsNil() { - return renderAfterFinish(valueResults[1].Interface().(error)) + if closeMethod.IsValid() { + defer closeMethod.Call(nil) + } + + for { + // Check for context cancellation (Ctrl+C) + select { + case <-ctx.Done(): + return false, finishJoin(wrap(ctx.Err())) + default: } - item = valueResults[0].Interface() - } else { - // EventStream: dereference pointer if needed - val := valueResults[0] - if val.Kind() == reflect.Ptr && !val.IsNil() { - item = val.Elem().Interface() - } else if val.Kind() == reflect.Ptr && val.IsNil() { - continue // skip nil events + + // Call Next() bool + nextResults := nextMethod.Call(nil) + stop() + if !nextResults[0].Bool() { + break + } + + // Call Value() — handles both: + // EventStream: Value() *T (1 return value) + // JsonLStream: Value() (T, error) (2 return values) + valueResults := valueMethod.Call(nil) + var item interface{} + if len(valueResults) == 2 { + // JsonLStream: check error — route through Error() for JSON formatting + if !valueResults[1].IsNil() { + return false, renderAfterFinish(wrap(valueResults[1].Interface().(error))) + } + item = valueResults[0].Interface() } else { - item = val.Interface() + // EventStream: dereference pointer if needed + val := valueResults[0] + if val.Kind() == reflect.Ptr && !val.IsNil() { + item = val.Elem().Interface() + } else if val.Kind() == reflect.Ptr && val.IsNil() { + continue // skip nil events + } else { + item = val.Interface() + } } - } - var raw []byte - if rawMethod.IsValid() { - raw = rawMethod.Call(nil)[0].Bytes() - } + var raw []byte + if rawMethod.IsValid() { + raw = rawMethod.Call(nil)[0].Bytes() + } - if len(valueResults) == 1 { - if body, isErr := streamErrorEvent(item, raw); isErr { - return renderAfterFinish(&StreamEventError{Body: string(body)}) + if len(valueResults) == 1 { + if body, isErr := streamErrorEvent(item, raw); isErr { + return false, renderAfterFinish(wrap(&StreamEventError{Body: string(body)})) + } } - } - var frame map[string]interface{} - if outcomeCfg != nil && streamFailure == nil { - frame, _ = decodeStreamFrame(raw) - streamFailure = outcomeStreamFailure(cmd, outcomeCfg, item, frame) + var frame map[string]interface{} + if outcomeCfg != nil && streamFailure == nil { + frame, _ = decodeStreamFrame(raw) + streamFailure = outcomeStreamFailure(cmd, outcomeCfg, item, frame) + } + if hops != nil { + if frame == nil { + frame, _ = decodeStreamFrame(raw) + } + hops.observe(frame) + } + item = wireStreamItem(item, raw, frame, format == "table" && jqExpr == "") + + if projector != nil { + if err := projector.emit(out, item); err != nil { + return false, finishJoin(err) + } + } else if err := outputOneItem(out, item, format, jqExpr, colorize, jqRaw, first); err != nil { + return false, err + } + first = false } - item = wireStreamItem(item, raw, frame, format == "table" && jqExpr == "") - if projector != nil { - if err := projector.emit(out, item); err != nil { - return finishJoin(err) + // Check for stream errors — route through Error() for consistent + // --output-format json behavior + if errMethod.IsValid() { + errResults := errMethod.Call(nil) + if len(errResults) > 0 && !errResults[0].IsNil() { + return false, renderAfterFinish(wrap(errResults[0].Interface().(error))) } - } else if err := outputOneItem(out, item, format, jqExpr, colorize, jqRaw, first); err != nil { - return err } - first = false + return false, nil } - // Check for stream errors — route through Error() for consistent - // --output-format json behavior - if errMethod.IsValid() { - errResults := errMethod.Call(nil) - if len(errResults) > 0 && !errResults[0].IsNil() { - return renderAfterFinish(errResults[0].Interface().(error)) + for hop := 0; ; hop++ { + fallback, err := drain(res) + if fallback { + stop() + if hop > 0 { + if finishErr := finishJoin(nil); finishErr != nil { + return finishErr + } + } + if hops != nil { + return hops.discrete(res) + } + return Result(cmd, res) } - } - if streamFailure != nil { - return renderAfterFinish(streamFailure) + if err != nil { + return err + } + if streamFailure != nil { + return renderAfterFinish(wrap(streamFailure)) + } + if hops == nil { + break + } + next, more, err := hops.next() + if err != nil { + return renderAfterFinish(err) + } + if !more { + break + } + res = next } if projector != nil { diff --git a/templates/templates/cli/includes/intents.ts b/templates/templates/cli/includes/intents.ts index 36cb9feaa..f6810e544 100644 --- a/templates/templates/cli/includes/intents.ts +++ b/templates/templates/cli/includes/intents.ts @@ -138,6 +138,9 @@ interface IntentCmdCtx { AsyncFailureHint: string; StreamMetadataJSON: string; OutcomeJSON: string; // terminal-state table carried via cobra annotation ("" = none) + ContinuationJSON: string; // continuation recipe carried via cobra annotation ("" = none) + ContinuationDefault: boolean; + ContinuationMaxHops: number; StreamSelect: string; // RFC 6901 pointer of output.stream.select ("" = none) StreamToggle: { Name: string; Key: string } | null; StreamKind: string; // "sse" | "jsonl" | "" — how the backing operation streams @@ -204,6 +207,9 @@ interface CLIOperationCtx { OperationID: string; StreamMetadataJSON: string; OutcomeJSON: string; + ContinuationJSON: string; + ContinuationDefault: boolean; + ContinuationMaxHops: number; StreamSelect: string; Flags: CLIOperationFlagCtx[]; BodyFlags: string[]; @@ -1355,6 +1361,9 @@ function collectIntentManifest(): IntentManifestCtx { StreamSelect: cmd.Output?.Stream?.Pointer || "", StreamMetadataJSON: streamMetadataJSON(cmd.Output?.Stream?.Metadata), OutcomeJSON: outcomeJSON(cmd.Output?.Outcome), + ContinuationJSON: continuationJSON(cmd.Continuation, found.op), + ContinuationDefault: cmd.Continuation?.Default ?? false, + ContinuationMaxHops: cmd.Continuation?.MaxHops ?? 0, StreamToggle: !dispatch && intentStreamKind(found.op).kind ? intentDefaultOnStreamToggle(flags, presetJSON) @@ -1419,6 +1428,64 @@ function outcomeJSON(outcome: CLICommandOutcome | null | undefined): string { }); } +// continuationJSON serializes a continuation recipe for the runtime. The +// linker cannot see pagination, so the paginated-operation conflict is +// rejected here. +function continuationJSON( + recipe: CLICommandContinuation | null | undefined, + op: Operation, +): string { + if (!recipe) { + return ""; + } + if (hasPagination(op)) { + throw new Error( + `x-speakeasy-cli-commands: operation "${op.OriginalID}" is paginated; continuation cannot be combined with pagination in v1`, + ); + } + return JSON.stringify({ + bind: recipe.Bind.map((binding) => ({ + ...(binding.Pointer ? { pointer: binding.Pointer } : {}), + ...(binding.StreamPointers?.length + ? { streamPointers: binding.StreamPointers } + : {}), + to: binding.To, + toPath: binding.ToPath, + ...(binding.Optional ? { optional: true } : {}), + ...(binding.Flag ? { flag: binding.Flag } : {}), + })), + ...(recipe.Merge?.length ? { merge: recipe.Merge } : {}), + ...(recipe.Sum?.length ? { sum: recipe.Sum } : {}), + maxHops: recipe.MaxHops, + }); +} + +function hasCliContinuations(): boolean { + const manifest = context.Global.AST.CLICommands as any; + return Boolean( + manifest?.Commands?.some((c: any) => c.Continuation) || + manifest?.Operations?.some((o: any) => o.Continuation), + ); +} +registerTemplateFunc("hasCliContinuations", hasCliContinuations); + +// Whether any command or operation entry runs a continuation over op, so the +// operation's run function needs the hop driver. +function operationHasContinuation(op: Operation): boolean { + const manifest = context.Global.AST.CLICommands as any; + const matches = (id: string | undefined) => + id === op.OriginalID || id === op.GetID(); + return Boolean( + manifest?.Commands?.some( + (c: any) => c.Continuation && matches(c.Continuation.OperationID), + ) || + manifest?.Operations?.some( + (o: any) => o.Continuation && matches(o.OperationID), + ), + ); +} +registerTemplateFunc("operationHasContinuation", operationHasContinuation); + function streamMetadataJSON( fields: CLICommandStreamMetadata[] | null | undefined, ): string { @@ -1630,6 +1697,9 @@ function getCLIOperationCtx(op: Operation): CLIOperationCtx | null { StreamSelect: declared.Output?.Stream?.Pointer || "", StreamMetadataJSON: streamMetadataJSON(declared.Output?.Stream?.Metadata), OutcomeJSON: outcomeJSON(declared.Output?.Outcome), + ContinuationJSON: continuationJSON(declared.Continuation, op), + ContinuationDefault: declared.Continuation?.Default ?? false, + ContinuationMaxHops: declared.Continuation?.MaxHops ?? 0, Flags: flags, BodyFlags: bodyFlags, CanonicalBodyFlag: bodyFlags[bodyFlags.length - 1] || "", diff --git a/templates/templates/cli/intentcmd.go.stmpl b/templates/templates/cli/intentcmd.go.stmpl index 6eedb5f8f..a3ecd534d 100644 --- a/templates/templates/cli/intentcmd.go.stmpl +++ b/templates/templates/cli/intentcmd.go.stmpl @@ -74,6 +74,9 @@ func InitIntent{{$c.FuncName}}(parent *cobra.Command) error { {{- if $c.OutcomeJSON}} "speakeasy_outcome": "{{escapeGoString $c.OutcomeJSON}}", {{- end}} + {{- if $c.ContinuationJSON}} + "speakeasy_continuation": "{{escapeGoString $c.ContinuationJSON}}", + {{- end}} {{- if $c.StreamSelect}} "speakeasy_stream_select": "{{escapeGoString $c.StreamSelect}}", {{- if $c.StreamMetadataJSON}} @@ -157,6 +160,10 @@ func InitIntent{{$c.FuncName}}(parent *cobra.Command) error { }) cmd.Flags().Bool("raw-response", false, "Print the raw API response instead of writing the {{$c.ArtifactKind}} to a file") {{- end}} + {{- if $c.ContinuationJSON}} + cmd.Flags().Bool("auto-continue", {{$c.ContinuationDefault}}, "Keep sending the request while the response asks to continue, and combine the responses") + cmd.Flags().Int("max-continuations", {{$c.ContinuationMaxHops}}, "Maximum number of responses to request when continuing automatically") + {{- end}} {{- if $c.AsyncJSON}} cmd.Flags().String("resume", "", "Resume polling an existing operation by ID instead of creating one") _ = flagutil.AnnotatePromptFlag(cmd, "resume", flagutil.PromptFlagSpec{Kind: "string"}) diff --git a/templates/templates/cli/opcmd.go.stmpl b/templates/templates/cli/opcmd.go.stmpl index 0e4624a92..bf1f85cab 100644 --- a/templates/templates/cli/opcmd.go.stmpl +++ b/templates/templates/cli/opcmd.go.stmpl @@ -49,6 +49,9 @@ func {{$initFuncName}}(parent *cobra.Command) error { {{- if and $cliOp $cliOp.OutcomeJSON}} parent.Annotations["speakeasy_outcome"] = "{{escapeGoString $cliOp.OutcomeJSON}}" {{- end}} + {{- if and $cliOp $cliOp.ContinuationJSON}} + parent.Annotations["speakeasy_continuation"] = "{{escapeGoString $cliOp.ContinuationJSON}}" + {{- end}} {{- if or $hasSecurity $op.Request (hasPagination $op) (hasBinaryResponse $op) (and $cliOp $cliOp.Flags)}} cmd := parent // register flags on the parent {{- end}} @@ -79,6 +82,9 @@ func {{$initFuncName}}(parent *cobra.Command) error { {{- if and $cliOp $cliOp.OutcomeJSON}} "speakeasy_outcome": "{{escapeGoString $cliOp.OutcomeJSON}}", {{- end}} + {{- if and $cliOp $cliOp.ContinuationJSON}} + "speakeasy_continuation": "{{escapeGoString $cliOp.ContinuationJSON}}", + {{- end}} }, } {{- end}} @@ -119,6 +125,10 @@ func {{$initFuncName}}(parent *cobra.Command) error { cmd.Flags().BoolP("all", "a", false, "Automatically paginate and fetch all results (streams NDJSON for JSON output)") cmd.Flags().Int("max-pages", 0, "Maximum number of pages to fetch when using --all (0 = no limit)") {{- end}} + {{- if and $cliOp $cliOp.ContinuationJSON}} + cmd.Flags().Bool("auto-continue", {{$cliOp.ContinuationDefault}}, "Keep sending the request while the response asks to continue, and combine the responses") + cmd.Flags().Int("max-continuations", {{$cliOp.ContinuationMaxHops}}, "Maximum number of responses to request when continuing automatically") + {{- end}} {{- if hasBinaryResponse $op}} cmd.Flags().String("output-file", "", "Save the response body to a file path (recommended for binary/file responses)") cmd.Flags().Bool("output-b64", false, "Encode binary response as base64 and print to stdout") @@ -345,6 +355,30 @@ func {{$runFuncName}}(cmd *cobra.Command, args []string) error { sdkOpts = append(sdkOpts, operations.WithSkipDeserialization()) } {{- end}} + {{- if and $op.Request (operationHasContinuation $op)}} + if output.ContinuationActive(cmd) { + hop := func() (interface{}, error) { + {{- if or (hasStreamingResponse $op) (not (operationHasDeferredBody $op))}} + stopProgress := output.StartRequestProgress(cmd) + defer stopProgress() + {{- end}} + res, err := s{{.Local.SubSDKAccessor}}.{{sanitizeMethodName $op}}({{templateSDKCallArgs $op $reqObjName}}) + if err != nil { + return nil, err + } + {{- if and (hasResponseTransform $op) (not (hasStreamingResponse $op))}} + output.DrainRawBody(res) + {{- end}} + return res, nil + } + {{- if hasStreamingResponse $op}} + if !client.IsDryRun(cmd) { + return output.ContinuationStream(cmd, &{{$reqObjName}}, hop, "{{getStreamingFieldName $op}}") + } + {{- end}} + return output.ContinuationResult(cmd, &{{$reqObjName}}, hop) + } + {{- end}} {{- if hasStreamingResponse $op}} // Streaming response — iterate events and output incrementally. // Skip streaming iteration in dry-run mode (synthetic response has no stream). diff --git a/templates/templates/cli/readme/agents.stmpl b/templates/templates/cli/readme/agents.stmpl index 07ee488e4..3e76f3f66 100644 --- a/templates/templates/cli/readme/agents.stmpl +++ b/templates/templates/cli/readme/agents.stmpl @@ -145,7 +145,7 @@ Outside agent mode, explicit JSON and `--jq` preserve the compatibility envelope {{- $carriers := cliErrorCarrierDisplayList}} -`error_type` is one of `authentication_error`, `authorization_error`, `not_found`, `validation_error`, `rate_limit_error`, `server_error`, `api_error`, `connection_error`, `protocol_error`, {{if cliEmitsServiceDisabledErrorType}}`service_disabled`, {{end}}{{if cliEmitsBillingDisabledErrorType}}`billing_disabled`, {{end}}`runtime_error`, `unsupported_error`, `async_failed`, `async_timeout`, `async_unknown_state`. {{if $carriers}}Classification reads the structured reason code at {{$carriers}} in the error body (resolved against the nested `error` object when the body has one) before HTTP status, so a declared credential reason sent with HTTP 400 is not mistaken for request validation. {{if cliErrorAnyCarrierDeclared}}`error_reason` carries the reason code found there, verbatim from a declared carrier when no declared rule matches it{{else}}`error_reason` carries the reason code found there when a declared rule matches it (an undeclared body reason is never promoted){{end}}; it is absent for status-only API errors.{{else}}Classification derives from the HTTP status and transport evidence; `error_reason` is absent for API errors.{{end}} Status-less local failures may use `CLI_VALIDATION`, `CLI_CONNECTION`, `CLI_PROTOCOL`, `CLI_RUNTIME`, `CLI_UNAVAILABLE`, `CLI_AUTHENTICATION`, the async polling reasons `CLI_ASYNC_FAILED`, `CLI_ASYNC_TIMEOUT`, and `CLI_ASYNC_UNKNOWN_STATE`, or `CLI_OUTCOME_FAILED` when a successful response reports a declared failure state; that envelope also carries `state` (the reported state) and, when the command declares an error pointer, `error` (the failure detail taken from the response). `hints` preserves server guidance first, adds the most specific local taxonomy guidance, then typed CLI and command-specific guidance, removing exact duplicates. `exit_code` is always the code for the final `error_type` shown in the envelope: 1 runtime, 2 usage, or 3 authentication/authorization. +`error_type` is one of `authentication_error`, `authorization_error`, `not_found`, `validation_error`, `rate_limit_error`, `server_error`, `api_error`, `connection_error`, `protocol_error`, {{if cliEmitsServiceDisabledErrorType}}`service_disabled`, {{end}}{{if cliEmitsBillingDisabledErrorType}}`billing_disabled`, {{end}}`runtime_error`, `unsupported_error`, `async_failed`, `async_timeout`, `async_unknown_state`. {{if $carriers}}Classification reads the structured reason code at {{$carriers}} in the error body (resolved against the nested `error` object when the body has one) before HTTP status, so a declared credential reason sent with HTTP 400 is not mistaken for request validation. {{if cliErrorAnyCarrierDeclared}}`error_reason` carries the reason code found there, verbatim from a declared carrier when no declared rule matches it{{else}}`error_reason` carries the reason code found there when a declared rule matches it (an undeclared body reason is never promoted){{end}}; it is absent for status-only API errors.{{else}}Classification derives from the HTTP status and transport evidence; `error_reason` is absent for API errors.{{end}} Status-less local failures may use `CLI_VALIDATION`, `CLI_CONNECTION`, `CLI_PROTOCOL`, `CLI_RUNTIME`, `CLI_UNAVAILABLE`, `CLI_AUTHENTICATION`, the async polling reasons `CLI_ASYNC_FAILED`, `CLI_ASYNC_TIMEOUT`, and `CLI_ASYNC_UNKNOWN_STATE`, or `CLI_OUTCOME_FAILED` when a successful response reports a declared failure state; that envelope also carries `state` (the reported state) and, when the command declares an error pointer, `error` (the failure detail taken from the response).{{if hasCliContinuations}} `CLI_CONTINUATION_LIMIT` means a command that continues a cut-off response reached its hop limit; that envelope carries `hops` and `continuation` (the resume values from the last response).{{end}} `hints` preserves server guidance first, adds the most specific local taxonomy guidance, then typed CLI and command-specific guidance, removing exact duplicates. `exit_code` is always the code for the final `error_type` shown in the envelope: 1 runtime, 2 usage, or 3 authentication/authorization. {{- if $c.Hints}} ```bash diff --git a/templates/templates/cli/tests/primary/continuation_additional_test.go.stmpl b/templates/templates/cli/tests/primary/continuation_additional_test.go.stmpl new file mode 100644 index 000000000..19d2fe9c1 --- /dev/null +++ b/templates/templates/cli/tests/primary/continuation_additional_test.go.stmpl @@ -0,0 +1,433 @@ +package tests + +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "{{getGolangPackage}}/internal/clierrors" +) + +type continuationReply struct { + status int + body string + events []string +} + +type continuationServer struct { + *httptest.Server + mu sync.Mutex + requests []map[string]interface{} +} + +// newContinuationServer answers the nth request (0-based) with reply(n) and +// records every request body so a test can check what each hop sent. +func newContinuationServer(t *testing.T, reply func(n int) continuationReply) *continuationServer { + t.Helper() + srv := &continuationServer{} + srv.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + raw, _ := io.ReadAll(r.Body) + var body map[string]interface{} + if err := json.Unmarshal(raw, &body); err != nil { + http.Error(w, "request body is not a JSON object", http.StatusBadRequest) + return + } + srv.mu.Lock() + n := len(srv.requests) + srv.requests = append(srv.requests, body) + srv.mu.Unlock() + + answer := reply(n) + if answer.status != 0 && answer.status != http.StatusOK { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(answer.status) + fmt.Fprint(w, answer.body) + return + } + switch r.URL.Path { + case "/continuation/compose": + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, answer.body) + case "/continuation/compose/stream": + if answer.events == nil { + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, answer.body) + return + } + w.Header().Set("Content-Type", "text/event-stream") + for _, event := range answer.events { + fmt.Fprintf(w, "data: %s\n\n", event) + } + fmt.Fprint(w, "data: [DONE]\n\n") + default: + http.NotFound(w, r) + } + })) + t.Cleanup(srv.Close) + return srv +} + +func (s *continuationServer) sent() []map[string]interface{} { + s.mu.Lock() + defer s.mu.Unlock() + return append([]map[string]interface{}(nil), s.requests...) +} + +func scriptedReplies(replies ...continuationReply) func(n int) continuationReply { + return func(n int) continuationReply { + if n < len(replies) { + return replies[n] + } + return continuationReply{status: http.StatusInternalServerError, body: `{"message":"unexpected extra request"}`} + } +} + +func widgetJSON(id, status, token, step string, inputUnits, outputUnits int) string { + fields := []string{fmt.Sprintf(`"status":%q`, status)} + if id != "" { + fields = append(fields, fmt.Sprintf(`"id":%q`, id)) + } + if token != "" { + fields = append(fields, fmt.Sprintf(`"resumeToken":%q`, token)) + } + if step != "" { + fields = append(fields, fmt.Sprintf(`"steps":[{"text":%q}]`, step)) + } + if inputUnits != 0 || outputUnits != 0 { + fields = append(fields, fmt.Sprintf(`"usage":{"inputUnits":%d,"outputUnits":%d}`, inputUnits, outputUnits)) + } + if status == "failed" { + fields = append(fields, `"error":{"code":"exhausted","message":"budget exhausted"}`) + } + return "{" + strings.Join(fields, ",") + "}" +} + +func discreteReply(id, status, token, step string, inputUnits, outputUnits int) continuationReply { + return continuationReply{body: widgetJSON(id, status, token, step, inputUnits, outputUnits)} +} + +func deltaEvent(text, token string) string { + if token == "" { + return fmt.Sprintf(`{"event_type":"delta","delta":{"text":%q}}`, text) + } + return fmt.Sprintf(`{"event_type":"delta","resumeToken":%q,"delta":{"text":%q}}`, token, text) +} + +func widgetEvent(id, status, token string) string { + return fmt.Sprintf(`{"event_type":"done","widget":%s}`, widgetJSON(id, status, token, "", 0, 0)) +} + +func composeArgs(extra ...string) []string { + return append([]string{"compose", "hello"}, extra...) +} + +func decodeJSONObject(t *testing.T, raw string) map[string]interface{} { + t.Helper() + var payload map[string]interface{} + require.NoError(t, json.Unmarshal([]byte(raw), &payload), raw) + return payload +} + +func stepTexts(payload map[string]interface{}) []string { + var texts []string + steps, _ := payload["steps"].([]interface{}) + for _, step := range steps { + if object, ok := step.(map[string]interface{}); ok { + text, _ := object["text"].(string) + texts = append(texts, text) + } + } + return texts +} + +func TestContinuationDiscrete(t *testing.T) { + threeHops := scriptedReplies( + discreteReply("w1", "more_required", "tok1", "a", 1, 2), + discreteReply("w2", "more_required", "tok2", "b", 3, 4), + discreteReply("w3", "completed", "", "c", 5, 6), + ) + + t.Run("single response", func(t *testing.T) { + srv := newContinuationServer(t, scriptedReplies(discreteReply("w1", "completed", "", "a", 1, 2))) + h, err := runOutcome(t, srv.URL, composeArgs("--output-format", "json")...) + require.Equal(t, clierrors.ExitOK, clierrors.ExitCode(err), h.GetStderr()) + sent := srv.sent() + require.Len(t, sent, 1) + assert.Equal(t, map[string]interface{}{"prompt": "hello"}, sent[0]) + payload := decodeJSONObject(t, h.GetStdout()) + assert.Equal(t, "completed", payload["status"]) + assert.Equal(t, []string{"a"}, stepTexts(payload)) + assert.Empty(t, h.GetStderr()) + }) + + t.Run("three responses are combined", func(t *testing.T) { + srv := newContinuationServer(t, threeHops) + h, err := runOutcome(t, srv.URL, composeArgs("--output-format", "json")...) + require.Equal(t, clierrors.ExitOK, clierrors.ExitCode(err), h.GetStderr()) + sent := srv.sent() + require.Len(t, sent, 3) + assert.Equal(t, map[string]interface{}{"prompt": "hello"}, sent[0]) + assert.Equal(t, map[string]interface{}{"prompt": "hello", "resumeToken": "tok1", "previousId": "w1"}, sent[1]) + assert.Equal(t, map[string]interface{}{"prompt": "hello", "resumeToken": "tok2", "previousId": "w2"}, sent[2]) + + payload := decodeJSONObject(t, h.GetStdout()) + assert.Equal(t, "w3", payload["id"], "fields other than merge and sum come from the last response") + assert.Equal(t, "completed", payload["status"]) + assert.Equal(t, []string{"a", "b", "c"}, stepTexts(payload)) + assert.Equal(t, map[string]interface{}{"inputUnits": float64(9), "outputUnits": float64(12)}, payload["usage"]) + assert.Empty(t, h.GetStderr()) + }) + + t.Run("sums beyond int64 stay exact", func(t *testing.T) { + srv := newContinuationServer(t, scriptedReplies( + continuationReply{body: `{"id":"w1","status":"more_required","resumeToken":"tok1","usage":{"inputUnits":9223372036854775807,"outputUnits":1}}`}, + continuationReply{body: `{"id":"w2","status":"completed","usage":{"inputUnits":1,"outputUnits":2}}`}, + )) + h, err := runOutcome(t, srv.URL, composeArgs("--output-format", "json")...) + require.Equal(t, clierrors.ExitOK, clierrors.ExitCode(err), h.GetStderr()) + assert.Contains(t, h.GetStdout(), `"usage":{"inputUnits":9223372036854775808,"outputUnits":3}`) + }) + + t.Run("three responses report the hop count", func(t *testing.T) { + srv := newContinuationServer(t, threeHops) + h, err := runOutcome(t, srv.URL, composeArgs()...) + require.Equal(t, clierrors.ExitOK, clierrors.ExitCode(err), h.GetStderr()) + assert.Contains(t, h.GetStderr(), "Continued across 3 responses.") + assert.NotContains(t, h.GetStderr(), "To continue") + }) + + t.Run("responses without an id omit the optional binding", func(t *testing.T) { + srv := newContinuationServer(t, scriptedReplies( + discreteReply("", "more_required", "tok1", "a", 0, 0), + discreteReply("", "completed", "", "b", 0, 0), + )) + h, err := runOutcome(t, srv.URL, composeArgs("--output-format", "json")...) + require.Equal(t, clierrors.ExitOK, clierrors.ExitCode(err), h.GetStderr()) + sent := srv.sent() + require.Len(t, sent, 2) + assert.Equal(t, map[string]interface{}{"prompt": "hello", "resumeToken": "tok1"}, sent[1]) + }) + + t.Run("resume values from flags reach the first request", func(t *testing.T) { + srv := newContinuationServer(t, scriptedReplies(discreteReply("w2", "completed", "", "b", 0, 0))) + h, err := runOutcome(t, srv.URL, composeArgs("--resume-token", "tok9", "--previous-id", "w9", "--output-format", "json")...) + require.Equal(t, clierrors.ExitOK, clierrors.ExitCode(err), h.GetStderr()) + sent := srv.sent() + require.Len(t, sent, 1) + assert.Equal(t, map[string]interface{}{"prompt": "hello", "resumeToken": "tok9", "previousId": "w9"}, sent[0]) + }) + + t.Run("a resumable success prints how to continue", func(t *testing.T) { + srv := newContinuationServer(t, scriptedReplies(discreteReply("w1", "budget_reached", "tok1", "a", 0, 0))) + h, err := runOutcome(t, srv.URL, composeArgs()...) + require.Equal(t, clierrors.ExitOK, clierrors.ExitCode(err), h.GetStderr()) + require.Len(t, srv.sent(), 1, "a success state never continues automatically") + assert.Contains(t, h.GetStderr(), `To continue, re-run the same command with --resume-token "tok1" --previous-id "w1".`) + }) + + t.Run("auto-continue disabled stops after the first response", func(t *testing.T) { + srv := newContinuationServer(t, threeHops) + h, err := runOutcome(t, srv.URL, composeArgs("--auto-continue=false")...) + require.Equal(t, clierrors.ExitOK, clierrors.ExitCode(err), h.GetStderr()) + require.Len(t, srv.sent(), 1) + assert.Contains(t, h.GetStderr(), `--resume-token "tok1"`) + }) + + t.Run("a failure after continuing carries the hop and resume values", func(t *testing.T) { + srv := newContinuationServer(t, scriptedReplies( + discreteReply("w1", "more_required", "tok1", "a", 0, 0), + discreteReply("w2", "failed", "tok2", "b", 0, 0), + )) + h, err := runOutcome(t, srv.URL, composeArgs("--output-format", "json")...) + require.Equal(t, clierrors.ExitRuntime, clierrors.ExitCode(err), h.GetStderr()) + require.Len(t, srv.sent(), 2) + envelope := outcomeErrorEnvelope(t, h.GetStderr()) + assert.Equal(t, "CLI_OUTCOME_FAILED", envelope["error_reason"]) + assert.Equal(t, "failed", envelope["state"]) + assert.Contains(t, envelope["message"], "budget exhausted") + assert.Equal(t, float64(2), envelope["hops"]) + assert.Equal(t, map[string]interface{}{"$.resumeToken": "tok2", "$.previousId": "w2"}, envelope["continuation"]) + assert.Equal(t, []string{"a", "b"}, stepTexts(decodeJSONObject(t, h.GetStdout()))) + }) + + t.Run("a continue state without a resume token is a protocol error", func(t *testing.T) { + srv := newContinuationServer(t, scriptedReplies(discreteReply("w1", "more_required", "", "a", 0, 0))) + h, err := runOutcome(t, srv.URL, composeArgs()...) + require.Equal(t, clierrors.ExitRuntime, clierrors.ExitCode(err), h.GetStderr()) + require.Len(t, srv.sent(), 1) + assert.Contains(t, h.GetStdout(), "more_required", "the response is still printed") + assert.Contains(t, h.GetStderr(), `reported state "more_required" but carried no $.resumeToken`) + }) + + t.Run("a continue state without a resume token is reported in json mode", func(t *testing.T) { + srv := newContinuationServer(t, scriptedReplies(discreteReply("w1", "more_required", "", "a", 0, 0))) + h, err := runOutcome(t, srv.URL, composeArgs("--output-format", "json")...) + require.Equal(t, clierrors.ExitRuntime, clierrors.ExitCode(err), h.GetStderr()) + assert.Equal(t, "more_required", decodeJSONObject(t, h.GetStdout())["status"]) + envelope := outcomeErrorEnvelope(t, h.GetStderr()) + assert.Equal(t, "CLI_PROTOCOL", envelope["error_reason"]) + assert.Equal(t, float64(1), envelope["hops"]) + }) + + t.Run("the hop limit stops an endless continuation", func(t *testing.T) { + srv := newContinuationServer(t, func(n int) continuationReply { + return discreteReply(fmt.Sprintf("w%d", n+1), "more_required", fmt.Sprintf("tok%d", n+1), "x", 0, 0) + }) + h, err := runOutcome(t, srv.URL, composeArgs("--max-continuations", "3", "--output-format", "json")...) + require.Equal(t, clierrors.ExitRuntime, clierrors.ExitCode(err), h.GetStderr()) + require.Len(t, srv.sent(), 3) + envelope := outcomeErrorEnvelope(t, h.GetStderr()) + assert.Equal(t, "CLI_CONTINUATION_LIMIT", envelope["error_reason"]) + assert.Equal(t, "runtime_error", envelope["error_type"]) + assert.Equal(t, float64(3), envelope["hops"]) + assert.Equal(t, map[string]interface{}{"$.resumeToken": "tok3", "$.previousId": "w3"}, envelope["continuation"]) + assert.Contains(t, fmt.Sprint(envelope["hints"]), `--resume-token "tok3"`) + assert.Len(t, stepTexts(decodeJSONObject(t, h.GetStdout())), 3) + }) + + t.Run("max-continuations must be positive", func(t *testing.T) { + srv := newContinuationServer(t, threeHops) + h, err := runOutcome(t, srv.URL, composeArgs("--max-continuations", "0")...) + require.Equal(t, clierrors.ExitUsage, clierrors.ExitCode(err), h.GetStderr()) + assert.Empty(t, srv.sent()) + }) + + t.Run("an error on a later hop carries the last resume values", func(t *testing.T) { + srv := newContinuationServer(t, scriptedReplies( + discreteReply("w1", "more_required", "tok1", "a", 0, 0), + continuationReply{status: http.StatusInternalServerError, body: `{"message":"upstream unavailable"}`}, + )) + h, err := runOutcome(t, srv.URL, composeArgs("--output-format", "json", "--no-retries")...) + require.NotEqual(t, clierrors.ExitOK, clierrors.ExitCode(err), h.GetStderr()) + require.Len(t, srv.sent(), 2) + envelope := outcomeErrorEnvelope(t, h.GetStderr()) + assert.Equal(t, "server_error", envelope["error_type"]) + assert.Equal(t, float64(1), envelope["hops"]) + assert.Equal(t, map[string]interface{}{"$.resumeToken": "tok1", "$.previousId": "w1"}, envelope["continuation"]) + }) + + t.Run("dry run sends nothing", func(t *testing.T) { + srv := newContinuationServer(t, threeHops) + h, err := runOutcome(t, srv.URL, composeArgs("--dry-run")...) + require.Equal(t, clierrors.ExitOK, clierrors.ExitCode(err), h.GetStderr()) + assert.Empty(t, srv.sent()) + }) +} + +func TestContinuationStream(t *testing.T) { + threeHops := scriptedReplies( + continuationReply{events: []string{deltaEvent("one ", "tok1"), widgetEvent("w1", "more_required", "")}}, + continuationReply{events: []string{deltaEvent("two ", ""), widgetEvent("w2", "more_required", "tok2")}}, + continuationReply{events: []string{deltaEvent("three", ""), widgetEvent("w3", "completed", "")}}, + ) + + t.Run("three responses stream as one output", func(t *testing.T) { + srv := newContinuationServer(t, threeHops) + h, err := runOutcome(t, srv.URL, "compose-stream", "hello") + require.Equal(t, clierrors.ExitOK, clierrors.ExitCode(err), h.GetStderr()) + assert.Equal(t, "one two three\n", h.GetStdout()) + sent := srv.sent() + require.Len(t, sent, 3) + assert.Equal(t, map[string]interface{}{"prompt": "hello"}, sent[0]) + assert.Equal(t, map[string]interface{}{"prompt": "hello", "resumeToken": "tok1", "previousId": "w1"}, sent[1]) + assert.Equal(t, map[string]interface{}{"prompt": "hello", "resumeToken": "tok2", "previousId": "w2"}, sent[2]) + assert.Contains(t, h.GetStderr(), "Continued across 3 responses.") + }) + + t.Run("json output carries every event of every response", func(t *testing.T) { + srv := newContinuationServer(t, threeHops) + h, err := runOutcome(t, srv.URL, "compose-stream", "hello", "--output-format", "json") + require.Equal(t, clierrors.ExitOK, clierrors.ExitCode(err), h.GetStderr()) + events := strings.Split(strings.TrimSpace(h.GetStdout()), "\n") + require.Len(t, events, 6, h.GetStdout()) + assert.Contains(t, events[5], `"completed"`) + }) + + t.Run("a stream that ends after a checkpoint is a connection error", func(t *testing.T) { + srv := newContinuationServer(t, scriptedReplies(continuationReply{events: []string{deltaEvent("one", "tok1")}})) + h, err := runOutcome(t, srv.URL, "compose-stream", "hello", "--output-format", "json") + require.Equal(t, clierrors.ExitRuntime, clierrors.ExitCode(err), h.GetStderr()) + require.Len(t, srv.sent(), 1) + envelope := outcomeErrorEnvelope(t, h.GetStderr()) + assert.Equal(t, "CLI_CONNECTION", envelope["error_reason"]) + assert.Equal(t, float64(1), envelope["hops"]) + assert.Equal(t, map[string]interface{}{"$.resumeToken": "tok1"}, envelope["continuation"]) + }) + + t.Run("a stream that ends while still pending is a connection error", func(t *testing.T) { + srv := newContinuationServer(t, scriptedReplies(continuationReply{events: []string{deltaEvent("one", "tok1"), widgetEvent("w1", "in_progress", "")}})) + h, err := runOutcome(t, srv.URL, "compose-stream", "hello", "--output-format", "json") + require.Equal(t, clierrors.ExitRuntime, clierrors.ExitCode(err), h.GetStderr()) + require.Len(t, srv.sent(), 1) + envelope := outcomeErrorEnvelope(t, h.GetStderr()) + assert.Equal(t, "CLI_CONNECTION", envelope["error_reason"]) + assert.Contains(t, envelope["message"], `still "in_progress"`) + assert.Equal(t, map[string]interface{}{"$.resumeToken": "tok1", "$.previousId": "w1"}, envelope["continuation"]) + }) + + t.Run("a json answer continues as a discrete response", func(t *testing.T) { + srv := newContinuationServer(t, scriptedReplies( + discreteReply("w1", "more_required", "tok1", "a", 0, 0), + discreteReply("w2", "completed", "", "b", 0, 0), + )) + h, err := runOutcome(t, srv.URL, "compose-stream", "hello", "--output-format", "json") + require.Equal(t, clierrors.ExitOK, clierrors.ExitCode(err), h.GetStderr()) + sent := srv.sent() + require.Len(t, sent, 2) + assert.Equal(t, map[string]interface{}{"prompt": "hello", "resumeToken": "tok1", "previousId": "w1"}, sent[1]) + assert.Equal(t, "w2", decodeJSONObject(t, h.GetStdout())["id"]) + }) + + t.Run("a stream continued by a json answer", func(t *testing.T) { + srv := newContinuationServer(t, scriptedReplies( + continuationReply{events: []string{deltaEvent("one", "tok1"), widgetEvent("w1", "more_required", "")}}, + discreteReply("w2", "completed", "", "b", 0, 0), + )) + h, err := runOutcome(t, srv.URL, "compose-stream", "hello") + require.Equal(t, clierrors.ExitOK, clierrors.ExitCode(err), h.GetStderr()) + require.Len(t, srv.sent(), 2) + assert.Contains(t, h.GetStdout(), "one") + assert.Contains(t, h.GetStdout(), "w2") + }) + + t.Run("a stream that ends without a checkpoint is left as is", func(t *testing.T) { + srv := newContinuationServer(t, scriptedReplies(continuationReply{events: []string{deltaEvent("one", "")}})) + h, err := runOutcome(t, srv.URL, "compose-stream", "hello") + require.Equal(t, clierrors.ExitOK, clierrors.ExitCode(err), h.GetStderr()) + assert.Equal(t, "one\n", h.GetStdout()) + require.Len(t, srv.sent(), 1) + }) + + t.Run("a failure on a later response", func(t *testing.T) { + srv := newContinuationServer(t, scriptedReplies( + continuationReply{events: []string{deltaEvent("one ", "tok1"), widgetEvent("w1", "more_required", "")}}, + continuationReply{events: []string{deltaEvent("two", ""), widgetEvent("w2", "failed", "tok2")}}, + )) + h, err := runOutcome(t, srv.URL, "compose-stream", "hello", "--output-format", "json") + require.Equal(t, clierrors.ExitRuntime, clierrors.ExitCode(err), h.GetStderr()) + require.Len(t, srv.sent(), 2) + envelope := outcomeErrorEnvelope(t, h.GetStderr()) + assert.Equal(t, "CLI_OUTCOME_FAILED", envelope["error_reason"]) + assert.Equal(t, float64(2), envelope["hops"]) + assert.Equal(t, map[string]interface{}{"$.resumeToken": "tok2", "$.previousId": "w2"}, envelope["continuation"]) + }) + + t.Run("the declared hop limit stops an endless stream", func(t *testing.T) { + srv := newContinuationServer(t, func(n int) continuationReply { + return continuationReply{events: []string{deltaEvent("x", fmt.Sprintf("tok%d", n+1)), widgetEvent("", "more_required", "")}} + }) + h, err := runOutcome(t, srv.URL, "compose-stream", "hello", "--output-format", "json") + require.Equal(t, clierrors.ExitRuntime, clierrors.ExitCode(err), h.GetStderr()) + require.Len(t, srv.sent(), 8) + envelope := outcomeErrorEnvelope(t, h.GetStderr()) + assert.Equal(t, "CLI_CONTINUATION_LIMIT", envelope["error_reason"]) + assert.Equal(t, float64(8), envelope["hops"]) + }) +} diff --git a/templates/templates/common/common/typeDefs.d.ts b/templates/templates/common/common/typeDefs.d.ts index b9aaf856a..5519aea7e 100644 --- a/templates/templates/common/common/typeDefs.d.ts +++ b/templates/templates/common/common/typeDefs.d.ts @@ -2418,6 +2418,7 @@ declare global { Flags: CLICommandInput[] | null; Presets: CLICommandPreset[] | null; Async?: CLICommandAsync | null; + Continuation?: CLICommandContinuation | null; Output: CLICommandOutput | null; Examples: CLICommandExample[] | null; Help?: CLICommandHelp | null; @@ -2429,10 +2430,42 @@ declare global { }[]; }; + type CLICommandContinuationBinding = { + From?: string; + Pointer?: string; + StreamFrom?: string[]; + StreamPointers?: string[]; + To: string; + ToPointer: string; + ToPath: string[]; + Optional?: boolean; + Flag?: string; + }; + + type CLICommandContinuation = { + Bind: CLICommandContinuationBinding[]; + MergeFrom?: string[]; + Merge?: string[]; + SumFrom?: string[]; + Sum?: string[]; + MaxHops: number; + Default: boolean; + OperationID?: string; + }; + + type CLIOperationDeclaration = { + OperationID: string; + Output?: CLICommandOutput | null; + Flags?: CLICommandInput[] | null; + Continuation?: CLICommandContinuation | null; + BodyRequired?: boolean; + }; + type CLICommandManifest = { Version: number; Categories?: string[]; Commands: CLICommand[]; + Operations?: CLIOperationDeclaration[] | null; }; type CLIErrorRule = { diff --git a/tests/overlays/primary/cli/overlay.yaml b/tests/overlays/primary/cli/overlay.yaml index 99300da9b..d9807cff7 100644 --- a/tests/overlays/primary/cli/overlay.yaml +++ b/tests/overlays/primary/cli/overlay.yaml @@ -157,7 +157,86 @@ actions: incomplete: success failed: failure cancelled: failure + continuationCompose: + output: + outcome: + statePointer: $.status + errorPointer: $.error + states: + in_progress: pending + more_required: continue + budget_reached: success + completed: success + failed: failure + continuation: + bind: + - from: $.resumeToken + to: $.resumeToken + - from: $.id + to: $.previousId + optional: true + merge: $.steps + sum: [$.usage.inputUnits, $.usage.outputUnits] commands: + compose: + category: Create + summary: Compose a widget, continuing automatically while the response asks for more (continuation inherited from the operation entry) + op: continuationCompose + args: + prompt: + to: $.prompt + variadic: true + summary: Prompt text + flags: + scenario: + to: $.scenario + summary: Test-service scenario to run + resume-token: + to: $.resumeToken + summary: Resume token from a previous response + previous-id: + to: $.previousId + summary: ID of the response that stopped + compose-stream: + category: Create + summary: Stream a widget composition across continued responses + op: continuationStream + args: + prompt: + to: $.prompt + variadic: true + summary: Prompt text + flags: + scenario: + to: $.scenario + summary: Test-service scenario to run + resume-token: + to: $.resumeToken + summary: Resume token from a previous response + output: + stream: + select: $.data.delta.text + outcome: + statePointer: $.status + errorPointer: $.error + streamStatePointer: $.data.widget.status + streamErrorPointer: $.data.widget.error + states: + in_progress: pending + more_required: continue + budget_reached: success + completed: success + failed: failure + continuation: + bind: + - from: $.resumeToken + streamFrom: [$.data.resumeToken, $.data.widget.resumeToken] + to: $.resumeToken + - from: $.id + streamFrom: $.data.widget.id + to: $.previousId + optional: true + maxHops: 8 outcome-check: category: Create summary: Fetch an interaction and fail when it ended in a failure state (output.outcome inherited from the operation entry) @@ -714,6 +793,50 @@ actions: x-speakeasy-sse-sentinel: "[DONE]" schema: $ref: '#/components/schemas/outcomeStreamEvent' + /continuation/compose: + post: + operationId: continuationCompose + tags: [continuation] + description: Each response returns only its own steps and usage, and reports more_required with a resume token while the composition is unfinished, so the CLI re-sends the same request with the resume values (continuation) and combines the responses. + servers: + - url: http://localhost:35456 + requestBody: + required: true + content: + application/json: + schema: + $ref: '#/components/schemas/continuationComposeRequest' + responses: + "200": + description: One response of the composition. + content: + application/json: + schema: + $ref: '#/components/schemas/continuationWidget' + /continuation/compose/stream: + post: + operationId: continuationStream + tags: [continuation] + description: Continued composition that answers either as an SSE stream, where the resume token arrives in stream events and the closing event carries the response's status, or as a single JSON document. + servers: + - url: http://localhost:35456 + requestBody: + required: true + content: + application/json: + schema: + $ref: '#/components/schemas/continuationComposeRequest' + responses: + "200": + description: Streamed composition events, or one JSON response. + content: + text/event-stream: + x-speakeasy-sse-sentinel: "[DONE]" + schema: + $ref: '#/components/schemas/continuationStreamEvent' + application/json: + schema: + $ref: '#/components/schemas/continuationWidget' - target: $.components.schemas update: outcomeInteraction: @@ -755,6 +878,63 @@ actions: type: array items: $ref: '#/components/schemas/outcomeError' + continuationComposeRequest: + type: object + required: [prompt] + properties: + prompt: + type: string + scenario: + type: string + resumeToken: + type: string + previousId: + type: string + continuationWidget: + type: object + required: [status] + properties: + id: + type: string + status: + type: string + enum: [in_progress, more_required, budget_reached, completed, failed] + resumeToken: + type: string + steps: + type: array + items: + type: object + properties: + text: + type: string + usage: + type: object + properties: + inputUnits: + type: integer + outputUnits: + type: integer + error: + $ref: '#/components/schemas/outcomeError' + continuationStreamEvent: + type: object + required: [data] + properties: + data: + type: object + properties: + event_type: + type: string + resumeToken: + type: string + delta: + type: object + properties: + text: + type: string + widget: + $ref: '#/components/schemas/continuationWidget' outcomeStreamEvent: type: object required: [data] diff --git a/zSDKs/sdk-cli/.speakeasy/gen.lock b/zSDKs/sdk-cli/.speakeasy/gen.lock index b040543bf..530a04158 100644 --- a/zSDKs/sdk-cli/.speakeasy/gen.lock +++ b/zSDKs/sdk-cli/.speakeasy/gen.lock @@ -274,7 +274,7 @@ trackedFiles: internal/output/binary.go: last_write_checksum: sha1:6fd77172cad1bd29dc9f157b931345f96a172223 internal/output/classify.go: - last_write_checksum: sha1:db7e072a1c68803b44a68ab0c0a620170bb5c6d1 + last_write_checksum: sha1:b21e34c493f5e773cdffcf0e3670af2c6e8bbffa internal/output/color.go: last_write_checksum: sha1:c6f41f9691f6209f480d153c15c3424d8ca5883d internal/output/errortable.go: @@ -298,7 +298,7 @@ trackedFiles: internal/output/progress.go: last_write_checksum: sha1:f8b839f9394602fe9c30feb3d7155af6235663be internal/output/streaming.go: - last_write_checksum: sha1:0bab1d6cbb6a5d6dca122c8d57a14467dd3f9237 + last_write_checksum: sha1:2a61d494008a17f1db0b7b1c3ae0f93fe73e50ca internal/output/streaming_test.go: last_write_checksum: sha1:ed4ffe1ebf27f5b32e69d0d2e4a377d47e1655c4 internal/output/streamproject.go: diff --git a/zSDKs/sdk-cli/internal/output/classify.go b/zSDKs/sdk-cli/internal/output/classify.go index b2f510e5a..30ee34a35 100644 --- a/zSDKs/sdk-cli/internal/output/classify.go +++ b/zSDKs/sdk-cli/internal/output/classify.go @@ -689,6 +689,7 @@ const ( ReasonCLIAsyncTimeout = "CLI_ASYNC_TIMEOUT" ReasonCLIAsyncUnknownState = "CLI_ASYNC_UNKNOWN_STATE" ReasonCLIOutcomeFailed = "CLI_OUTCOME_FAILED" + ReasonCLIContinuationLimit = "CLI_CONTINUATION_LIMIT" ) type cliReasonError struct { @@ -711,7 +712,7 @@ func normalizeCLIReason(reason string) string { case ReasonCLIValidation, ReasonCLIConnection, ReasonCLIProtocol, ReasonCLIRuntime, ReasonCLIUnavailable, ReasonCLIAuthentication, ReasonCLIAsyncFailed, ReasonCLIAsyncTimeout, ReasonCLIAsyncUnknownState, - ReasonCLIOutcomeFailed: + ReasonCLIOutcomeFailed, ReasonCLIContinuationLimit: return reason } return ReasonCLIRuntime diff --git a/zSDKs/sdk-cli/internal/output/streaming.go b/zSDKs/sdk-cli/internal/output/streaming.go index c5b89175d..5809255cd 100644 --- a/zSDKs/sdk-cli/internal/output/streaming.go +++ b/zSDKs/sdk-cli/internal/output/streaming.go @@ -32,6 +32,22 @@ func StreamResult(cmd *cobra.Command, res interface{}, streamFieldName string, s } } } + return streamResult(cmd, res, streamFieldName, nil, stop) +} + +// streamHops continues a stream into the next hop's stream when one ends, +// so every hop's events reach one projector and one output sequence. +type streamHops interface { + observe(frame map[string]interface{}) + wrap(err error) error + next() (interface{}, bool, error) + discrete(res interface{}) error +} + +func streamResult(cmd *cobra.Command, res interface{}, streamFieldName string, hops streamHops, stop func()) error { + if stop == nil { + stop = func() {} + } defer stop() if flagutil.DidDryRunRequest(cmd) { return nil @@ -44,48 +60,6 @@ func StreamResult(cmd *cobra.Command, res interface{}, streamFieldName string, s jqExpr, _ := flagutil.GetStringFlag(cmd, "jq") jqRaw := jqRawOutput(cmd) - // Extract the stream field from the response envelope via reflection - v := reflect.ValueOf(res) - if v.Kind() == reflect.Ptr { - if v.IsNil() { - return nil - } - v = v.Elem() - } - if v.Kind() != reflect.Struct { - // Fallback: not a struct envelope — output as regular result - stop() - return Result(cmd, res) - } - streamVal := v.FieldByName(streamFieldName) - if !streamVal.IsValid() { - // Fallback: stream field not found — output as regular result - stop() - return Result(cmd, res) - } - // Guard IsNil: only call on nillable kinds. Fall back to Result() for - // mixed-content operations where non-stream content may still exist. - if isNillable(streamVal) && streamVal.IsNil() { - stop() - return Result(cmd, res) - } - - // Validate streaming interface: must have Next and Value methods - nextMethod := streamVal.MethodByName("Next") - 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 - stop() - return Result(cmd, res) - } - - if closeMethod.IsValid() { - defer closeMethod.Call(nil) - } - projector := newStreamProjector(cmd) outcomeCfg := activeOutcome(cmd) if outcomeCfg != nil && outcomeCfg.StreamStatePointer == "" { @@ -113,84 +87,166 @@ func StreamResult(cmd *cobra.Command, res interface{}, streamFieldName string, s } return rendered } + wrap := func(e error) error { + if hops != nil { + return hops.wrap(e) + } + return e + } first := true - for { - // Check for context cancellation (Ctrl+C) - select { - case <-ctx.Done(): - return finishJoin(ctx.Err()) - default: - } - - // Call Next() bool - nextResults := nextMethod.Call(nil) - stop() - if !nextResults[0].Bool() { - break + // drain writes one response's events. fallback reports a response that + // carries no stream, which is rendered as a regular result instead. + drain := func(res interface{}) (fallback bool, err error) { + // Extract the stream field from the response envelope via reflection + v := reflect.ValueOf(res) + if v.Kind() == reflect.Ptr { + if v.IsNil() { + return false, nil + } + v = v.Elem() + } + if v.Kind() != reflect.Struct { + return true, nil + } + streamVal := v.FieldByName(streamFieldName) + if !streamVal.IsValid() { + return true, nil + } + // Guard IsNil: only call on nillable kinds. Fall back to Result() for + // mixed-content operations where non-stream content may still exist. + if isNillable(streamVal) && streamVal.IsNil() { + return true, nil + } + + // Validate streaming interface: must have Next and Value methods + nextMethod := streamVal.MethodByName("Next") + valueMethod := streamVal.MethodByName("Value") + errMethod := streamVal.MethodByName("Err") + closeMethod := streamVal.MethodByName("Close") + rawMethod := streamVal.MethodByName("RawValue") + if !nextMethod.IsValid() || !valueMethod.IsValid() { + return true, nil } - // Call Value() — handles both: - // EventStream: Value() *T (1 return value) - // JsonLStream: Value() (T, error) (2 return values) - valueResults := valueMethod.Call(nil) - var item interface{} - if len(valueResults) == 2 { - // JsonLStream: check error — route through Error() for JSON formatting - if !valueResults[1].IsNil() { - return renderAfterFinish(valueResults[1].Interface().(error)) + if closeMethod.IsValid() { + defer closeMethod.Call(nil) + } + + for { + // Check for context cancellation (Ctrl+C) + select { + case <-ctx.Done(): + return false, finishJoin(wrap(ctx.Err())) + default: } - item = valueResults[0].Interface() - } else { - // EventStream: dereference pointer if needed - val := valueResults[0] - if val.Kind() == reflect.Ptr && !val.IsNil() { - item = val.Elem().Interface() - } else if val.Kind() == reflect.Ptr && val.IsNil() { - continue // skip nil events + + // Call Next() bool + nextResults := nextMethod.Call(nil) + stop() + if !nextResults[0].Bool() { + break + } + + // Call Value() — handles both: + // EventStream: Value() *T (1 return value) + // JsonLStream: Value() (T, error) (2 return values) + valueResults := valueMethod.Call(nil) + var item interface{} + if len(valueResults) == 2 { + // JsonLStream: check error — route through Error() for JSON formatting + if !valueResults[1].IsNil() { + return false, renderAfterFinish(wrap(valueResults[1].Interface().(error))) + } + item = valueResults[0].Interface() } else { - item = val.Interface() + // EventStream: dereference pointer if needed + val := valueResults[0] + if val.Kind() == reflect.Ptr && !val.IsNil() { + item = val.Elem().Interface() + } else if val.Kind() == reflect.Ptr && val.IsNil() { + continue // skip nil events + } else { + item = val.Interface() + } } - } - var raw []byte - if rawMethod.IsValid() { - raw = rawMethod.Call(nil)[0].Bytes() - } + var raw []byte + if rawMethod.IsValid() { + raw = rawMethod.Call(nil)[0].Bytes() + } - if len(valueResults) == 1 { - if body, isErr := streamErrorEvent(item, raw); isErr { - return renderAfterFinish(&StreamEventError{Body: string(body)}) + if len(valueResults) == 1 { + if body, isErr := streamErrorEvent(item, raw); isErr { + return false, renderAfterFinish(wrap(&StreamEventError{Body: string(body)})) + } } - } - var frame map[string]interface{} - if outcomeCfg != nil && streamFailure == nil { - frame, _ = decodeStreamFrame(raw) - streamFailure = outcomeStreamFailure(cmd, outcomeCfg, item, frame) + var frame map[string]interface{} + if outcomeCfg != nil && streamFailure == nil { + frame, _ = decodeStreamFrame(raw) + streamFailure = outcomeStreamFailure(cmd, outcomeCfg, item, frame) + } + if hops != nil { + if frame == nil { + frame, _ = decodeStreamFrame(raw) + } + hops.observe(frame) + } + item = wireStreamItem(item, raw, frame, format == "table" && jqExpr == "") + + if projector != nil { + if err := projector.emit(out, item); err != nil { + return false, finishJoin(err) + } + } else if err := outputOneItem(out, item, format, jqExpr, colorize, jqRaw, first); err != nil { + return false, err + } + first = false } - item = wireStreamItem(item, raw, frame, format == "table" && jqExpr == "") - if projector != nil { - if err := projector.emit(out, item); err != nil { - return finishJoin(err) + // Check for stream errors — route through Error() for consistent + // --output-format json behavior + if errMethod.IsValid() { + errResults := errMethod.Call(nil) + if len(errResults) > 0 && !errResults[0].IsNil() { + return false, renderAfterFinish(wrap(errResults[0].Interface().(error))) } - } else if err := outputOneItem(out, item, format, jqExpr, colorize, jqRaw, first); err != nil { - return err } - first = false + return false, nil } - // Check for stream errors — route through Error() for consistent - // --output-format json behavior - if errMethod.IsValid() { - errResults := errMethod.Call(nil) - if len(errResults) > 0 && !errResults[0].IsNil() { - return renderAfterFinish(errResults[0].Interface().(error)) + for hop := 0; ; hop++ { + fallback, err := drain(res) + if fallback { + stop() + if hop > 0 { + if finishErr := finishJoin(nil); finishErr != nil { + return finishErr + } + } + if hops != nil { + return hops.discrete(res) + } + return Result(cmd, res) } - } - if streamFailure != nil { - return renderAfterFinish(streamFailure) + if err != nil { + return err + } + if streamFailure != nil { + return renderAfterFinish(wrap(streamFailure)) + } + if hops == nil { + break + } + next, more, err := hops.next() + if err != nil { + return renderAfterFinish(err) + } + if !more { + break + } + res = next } if projector != nil {