diff --git a/docs/architecture/README.md b/docs/architecture/README.md index f5303ad03b..245be44513 100644 --- a/docs/architecture/README.md +++ b/docs/architecture/README.md @@ -7,27 +7,27 @@ the agent processes. ## Resource ownership -| Resource | Owner | Purpose | -| --- | --- | --- | -| `Harness` | Kubernetes (`kagent.dev/v1alpha3`) | Runtime implementation, workload, credentials, capacity, and admission policy | -| `AgentTemplate` | Kubernetes (`kagent.dev/v1alpha3`) | Portable agent behavior: model, prompt, tools, skills, and plugins | -| prepared revision | PostgreSQL and ate-api | Immutable compiled runtime input and its Substrate ActorTemplate | -| `AgentInstance` | PostgreSQL, exposed by gRPC | Ephemeral compute identity and lifecycle | -| A2A context, task, and events | PostgreSQL, exposed by A2A | Durable interaction and audit history | -| checkpoint | PostgreSQL plus a Substrate snapshot tag | Immutable, named restart boundary | -| Actor and durable directory | Substrate | Process lifecycle and private runtime state | +| Resource | Owner | Purpose | +| ----------------------------- | ---------------------------------------- | ----------------------------------------------------------------------------- | +| `Harness` | Kubernetes (`kagent.dev/v1alpha3`) | Runtime implementation, workload, credentials, capacity, and admission policy | +| `AgentTemplate` | Kubernetes (`kagent.dev/v1alpha3`) | Portable agent behavior: model, prompt, tools, skills, and plugins | +| prepared revision | PostgreSQL and ate-api | Immutable compiled runtime input and its Substrate ActorTemplate | +| `AgentInstance` | PostgreSQL, exposed by gRPC | Ephemeral compute identity and lifecycle | +| A2A context, task, and events | PostgreSQL, exposed by A2A | Durable interaction and audit history | +| checkpoint | PostgreSQL plus a Substrate snapshot tag | Immutable, named restart boundary | +| Actor and durable directory | Substrate | Process lifecycle and private runtime state | `AgentInstance` is not a Kubernetes resource. A2A owns public interaction semantics; kagent does not maintain a parallel session or task API. ## Public surfaces -| Surface | Role | -| --- | --- | -| Kubernetes API | Author Harnesses, AgentTemplates, models, prompts, and remote MCP servers | -| gRPC / gRPC-Web | Manage AgentInstances, sharing, checkpoints, and control-plane reads | -| A2A | Invoke agents and manage durable tasks and streams | -| MCP | Discover, invoke, checkpoint, and fork AgentInstances through A2A semantics | +| Surface | Role | +| --------------- | --------------------------------------------------------------------------- | +| Kubernetes API | Author Harnesses, AgentTemplates, models, prompts, and remote MCP servers | +| gRPC / gRPC-Web | Manage AgentInstances, sharing, checkpoints, and control-plane reads | +| A2A | Invoke agents and manage durable tasks and streams | +| MCP | Discover, invoke, checkpoint, and fork AgentInstances through A2A semantics | ## End-to-end flow @@ -77,6 +77,7 @@ quiescence. It reaches Actors through the private runtime network. - [Human in the loop](human-in-the-loop.md) - [Prompt resolution](prompt-templates.md) - [Telemetry](telemetry.md) +- [Structured output](structured-output.md) The documents describe implemented behavior. Deferred work, including full cross-AgentInstance delegation and Dedicated agents, belongs in the diff --git a/docs/architecture/structured-output.md b/docs/architecture/structured-output.md new file mode 100644 index 0000000000..2ce837865f --- /dev/null +++ b/docs/architecture/structured-output.md @@ -0,0 +1,81 @@ +# Structured Output + +An `AgentTemplate` can declare a JSON Schema for its successful terminal +response. The schema is either written inline in `spec.outputSchema` or stored +as JSON in a same-namespace ConfigMap selected by `spec.outputSchemaFrom`. The +two fields are mutually exclusive. + +```yaml +apiVersion: kagent.dev/v1alpha3 +kind: AgentTemplate +metadata: + name: data-extractor +spec: + modelConfig: + name: default-model-config + outputSchema: + type: object + properties: + status: + type: string + payload: + type: object + required: [status, payload] + additionalProperties: false +``` + +The translator resolves ConfigMap references, validates the portable schema +profile, and records a canonical schema and digest in the immutable prepared +revision. Invalid schemas and unsupported Harnesses fail compilation. The +kagent Harness supports structured output in its Go ADK runtime; +Codex, Claude, and BYO Harnesses do not currently support it. + +The contract belongs to the public root agent. It is not inherited by local +child agents or agent tools. If a child template is compiled and invoked as a +root agent, its own schema applies to that independent revision. + +## A2A response contract + +The runtime suppresses partial root-answer fragments, validates the complete +value, and publishes the successful result as one A2A `DataPart` with media +type `application/json`. The part also carries the canonical schema digest in +`kagent.dev/a2a/output-schema-sha256`. The generated Agent Card advertises +`application/json` as its default output mode. + +Progress updates, tool events, approval requests, and input-required messages +keep their existing representations. Consequently, a streaming client that +only needs the answer reads the last content-bearing artifact before the Task +reaches `TASK_STATE_COMPLETED`. The gateway persists and relays these events +without interpreting the structured payload. + +Structured output has no text fallback. Provider refusal, incomplete output, +invalid JSON, or schema validation failure fails the Task. Invalid model output +is not published or included in errors because it may contain sensitive data. + +## Portable schema profile + +Schemas must have an object root. The portable profile supports primitive JSON +types, nested objects and arrays, `properties`, `required`, +`additionalProperties`, `items`, `enum`, `const`, `anyOf`, titles and +descriptions, and non-recursive local references through `$defs`. + +External or recursive references, conditional schemas, tuple arrays, and other +unsupported keywords are rejected during compilation rather than silently +weakened for a provider. A provider or model can still reject structured output +at runtime; in that case the Task fails. + +The serialized schema is limited to 64 KiB. Conversion to the Go ADK schema is +also limited to 32 levels and 1,000 expanded schema nodes; these bounds include +nodes expanded through local `$defs` references. Schemas that exceed any limit +fail compilation. + +The semantic resolution code is in +[`go/core/internal/translator/outputschema.go`](../../go/core/internal/translator/outputschema.go). +Runtime validation and A2A conversion are owned by the Go kagent ADK executor. + +## References + +- [OpenAI Structured Outputs](https://platform.openai.com/docs/guides/structured-outputs) +- [Gemini structured outputs](https://ai.google.dev/gemini-api/docs/structured-output) +- [Claude structured outputs](https://platform.claude.com/docs/en/build-with-claude/structured-outputs) +- [Amazon Bedrock structured outputs](https://docs.aws.amazon.com/bedrock/latest/userguide/structured-output.html) diff --git a/go/adk/cmd/main.go b/go/adk/cmd/main.go index 530c6a39f9..03f772b928 100644 --- a/go/adk/cmd/main.go +++ b/go/adk/cmd/main.go @@ -195,13 +195,18 @@ func main() { } stream := agentConfig.GetStream() - executor := a2a.NewKAgentExecutor(a2a.KAgentExecutorConfig{ + executor, err := a2a.NewKAgentExecutor(a2a.KAgentExecutorConfig{ RunnerConfig: runnerConfig, SessionService: sessionService, Stream: stream, AppName: appName, Logger: logger, + Output: agentConfig.Output, }) + if err != nil { + logger.Error("failed to create A2A executor", "error", err) + os.Exit(1) + } // Build the agent card. if agentCard == nil { diff --git a/go/adk/pkg/a2a/executor.go b/go/adk/pkg/a2a/executor.go index eeee7b00ae..5e0c597983 100644 --- a/go/adk/pkg/a2a/executor.go +++ b/go/adk/pkg/a2a/executor.go @@ -2,8 +2,10 @@ package a2a import ( "context" + "encoding/json" "fmt" "iter" + "slices" "strings" "time" @@ -11,10 +13,12 @@ import ( a2atype "github.com/a2aproject/a2a-go/v2/a2a" "github.com/a2aproject/a2a-go/v2/a2asrv" + "github.com/google/jsonschema-go/jsonschema" "github.com/kagent-dev/kagent/go/adk/pkg/auth" "github.com/kagent-dev/kagent/go/adk/pkg/models" "github.com/kagent-dev/kagent/go/adk/pkg/telemetry" apia2a "github.com/kagent-dev/kagent/go/api/a2a" + apiadk "github.com/kagent-dev/kagent/go/api/adk" "github.com/kagent-dev/kagent/go/pkg/tracing" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/trace" @@ -22,6 +26,7 @@ import ( "google.golang.org/adk/v2/runner" "google.golang.org/adk/v2/server/adka2a/v2" adksession "google.golang.org/adk/v2/session" + "google.golang.org/genai" ) const ( @@ -35,21 +40,33 @@ type KAgentExecutorConfig struct { Stream bool AppName string Logger *slog.Logger + Output *apiadk.OutputConfig } // KAgentExecutor keeps kagent's request/session glue around the upstream ADK // A2A executor. Event conversion and artifact streaming are delegated to ADK. type KAgentExecutor struct { - builtin a2asrv.AgentExecutor - sessionService adksession.Service - appName string - logger *slog.Logger + builtin a2asrv.AgentExecutor + sessionService adksession.Service + appName string + logger *slog.Logger + structuredOutputEnabled bool +} + +type structuredOutput struct { + schema *jsonschema.Resolved + sha256 string } var _ a2asrv.AgentExecutor = (*KAgentExecutor)(nil) -// NewKAgentExecutor creates a KAgentExecutor from config. -func NewKAgentExecutor(cfg KAgentExecutorConfig) *KAgentExecutor { +// NewKAgentExecutor creates a KAgentExecutor from config. It returns an error if +// the configured output schema cannot be prepared for runtime validation. +func NewKAgentExecutor(cfg KAgentExecutorConfig) (*KAgentExecutor, error) { + output, err := resolveStructuredOutput(cfg.Output) + if err != nil { + return nil, err + } var runConfig adkagent.RunConfig if cfg.Stream { runConfig.StreamingMode = adkagent.StreamingModeSSE @@ -58,11 +75,18 @@ func NewKAgentExecutor(cfg KAgentExecutorConfig) *KAgentExecutor { if cfg.SessionService != nil { runnerConfig.SessionService = cfg.SessionService } + if runnerConfig.Agent == nil { + return nil, fmt.Errorf("root agent is required") + } + rootName := runnerConfig.Agent.Name() + if rootName == "" { + return nil, fmt.Errorf("root agent name is required") + } builtin := adka2a.NewExecutor(adka2a.ExecutorConfig{ RunnerConfig: runnerConfig, RunConfig: runConfig, A2APartConverter: a2aPartConverter, - GenAIPartConverter: genAIPartConverter, + GenAIPartConverter: structuredOutputPartConverter(output, rootName), AfterEventCallback: func(ctx adka2a.ExecutorContext, event *adksession.Event, processed *a2atype.TaskArtifactUpdateEvent) error { if event.InvocationID != "" { trace.SpanFromContext(ctx).SetAttributes(attribute.String("gcp.vertex.agent.invocation_id", event.InvocationID)) @@ -77,17 +101,105 @@ func NewKAgentExecutor(cfg KAgentExecutorConfig) *KAgentExecutor { } processed.Artifact.SetMeta(apia2a.TimelinePositionMetadataKey, position.UTC().Format(time.RFC3339Nano)) } - return nil + return transformStructuredOutput(output, rootName, event, processed) }, OutputMode: adka2a.OutputArtifactPerEvent, }) return &KAgentExecutor{ - builtin: builtin, - sessionService: runnerConfig.SessionService, - appName: cfg.AppName, - logger: cfg.Logger.With("component", "kagent-executor"), + builtin: builtin, + sessionService: runnerConfig.SessionService, + appName: cfg.AppName, + logger: cfg.Logger.With("component", "kagent-executor"), + structuredOutputEnabled: output != nil, + }, nil +} + +// structuredOutputPartConverter drops partial root output before upstream ADK +// creates an A2A artifact. A structured response is not useful until it is a +// complete JSON value, and publishing fragments could persist invalid JSON. +// Events from tools and sub-agents retain the normal converter behavior. +func structuredOutputPartConverter(output *structuredOutput, rootName string) adka2a.GenAIPartConverter { + return func(ctx context.Context, event *adksession.Event, part *genai.Part) (*a2atype.Part, error) { + if output != nil && event != nil && event.Author == rootName && event.Partial { + return nil, nil + } + return genAIPartConverter(ctx, event, part) + } +} + +func transformStructuredOutput(output *structuredOutput, rootName string, event *adksession.Event, processed *a2atype.TaskArtifactUpdateEvent) error { + if output == nil || event.Author != rootName { + return nil + } + // ADK treats interruption and skip-summarization events as final responses so + // the current agent loop stops. They are not final structured results. + if !event.IsFinalResponse() || len(event.LongRunningToolIDs) > 0 || event.Actions.SkipSummarization { + return nil + } + switch event.FinishReason { + case "", genai.FinishReasonUnspecified, genai.FinishReasonStop: + default: + return fmt.Errorf("output_validation_failed: root agent did not complete structured output") + } + value, err := finalStructuredOutput(event) + if err != nil { + return err + } + if err := output.schema.Validate(value); err != nil { + return fmt.Errorf("output_validation_failed: root agent output does not conform to its schema") } + if processed == nil || processed.Artifact == nil { + return fmt.Errorf("output_validation_failed: root agent produced no result artifact") + } + part := apia2a.NewStructuredOutputPart(value, output.sha256) + processed.Artifact.Parts = a2atype.ContentParts{part} + return nil +} + +// finalStructuredOutput returns ADK's parsed output when available. Normal A2A +// chat execution does not run the workflow-node wrapper that populates +// Event.Output, so in that path the final model text is decoded here before it +// is validated against the canonical JSON Schema. +func finalStructuredOutput(event *adksession.Event) (any, error) { + if event.Output != nil { + return event.Output, nil + } + if event.Content == nil { + return nil, fmt.Errorf("output_validation_failed: root agent produced no structured value") + } + var text strings.Builder + for _, part := range event.Content.Parts { + if part == nil || part.Thought { + continue + } + text.WriteString(part.Text) + } + if strings.TrimSpace(text.String()) == "" { + return nil, fmt.Errorf("output_validation_failed: root agent produced no structured value") + } + var value any + if err := json.Unmarshal([]byte(text.String()), &value); err != nil { + return nil, fmt.Errorf("output_validation_failed: root agent output is not valid JSON") + } + return value, nil +} + +// resolveStructuredOutput compiles the configured JSON Schema once so every +// execution can validate its final value with the same immutable validator. +func resolveStructuredOutput(config *apiadk.OutputConfig) (*structuredOutput, error) { + if config == nil { + return nil, nil + } + var schema jsonschema.Schema + if err := json.Unmarshal(config.JSONSchema, &schema); err != nil { + return nil, fmt.Errorf("decode output schema: %w", err) + } + resolved, err := schema.Resolve(nil) + if err != nil { + return nil, fmt.Errorf("resolve output schema: %w", err) + } + return &structuredOutput{schema: resolved, sha256: config.SHA256}, nil } // UserIDCallInterceptor returns an a2asrv.CallInterceptor that extracts the @@ -187,7 +299,26 @@ func (e *KAgentExecutor) Execute(ctx context.Context, reqCtx *a2asrv.ExecutorCon reqCtx.Message = resumeMessage } + structuredResultEmitted := false for event, err := range e.builtin.Execute(ctx, reqCtx) { + // Mark that a structured result has been emitted + if update, ok := event.(*a2atype.TaskArtifactUpdateEvent); ok && update.Artifact != nil && + slices.ContainsFunc(update.Artifact.Parts, apia2a.IsStructuredOutputPart) { + structuredResultEmitted = true + } + // If the event is a task status update event and the status is completed, + // but no structured result has been emitted, mark the status as failed with a message + // indicating that the root agent produced no result artifact. + if update, ok := event.(*a2atype.TaskStatusUpdateEvent); ok && + err == nil && e.structuredOutputEnabled && !structuredResultEmitted && + update.Status.State == a2atype.TaskStateCompleted { + update.Status.State = a2atype.TaskStateFailed + update.Status.Message = a2atype.NewMessageForTask( + a2atype.MessageRoleAgent, + update, + a2atype.NewTextPart("output_validation_failed: root agent produced no result artifact"), + ) + } // If the event is a task status update event and the status is input required, build the HITL status message if update, ok := event.(*a2atype.TaskStatusUpdateEvent); ok && update.Status.State == a2atype.TaskStateInputRequired && update.Status.Message != nil { diff --git a/go/adk/pkg/a2a/executor_test.go b/go/adk/pkg/a2a/executor_test.go index be0bc95cc1..a088afa66a 100644 --- a/go/adk/pkg/a2a/executor_test.go +++ b/go/adk/pkg/a2a/executor_test.go @@ -3,7 +3,9 @@ package a2a import ( "context" "iter" + "maps" "slices" + "strings" "testing" "log/slog" @@ -12,6 +14,7 @@ import ( "github.com/a2aproject/a2a-go/v2/a2asrv" "github.com/kagent-dev/kagent/go/adk/pkg/auth" apia2a "github.com/kagent-dev/kagent/go/api/a2a" + apiadk "github.com/kagent-dev/kagent/go/api/adk" "go.opentelemetry.io/otel" sdktrace "go.opentelemetry.io/otel/sdk/trace" "go.opentelemetry.io/otel/sdk/trace/tracetest" @@ -19,6 +22,7 @@ import ( "google.golang.org/adk/v2/model" "google.golang.org/adk/v2/runner" adksession "google.golang.org/adk/v2/session" + "google.golang.org/adk/v2/tool/toolconfirmation" "google.golang.org/genai" ) @@ -204,6 +208,326 @@ func TestKAgentExecutor_PreservesContentBearingLastChunk(t *testing.T) { } } +func TestTransformStructuredOutputParsesFinalTextWhenADKOutputIsUnset(t *testing.T) { + output, err := resolveStructuredOutput(&apiadk.OutputConfig{ + JSONSchema: []byte(`{"type":"object","properties":{"answer":{"type":"integer"}},"required":["answer"],"additionalProperties":false}`), + SHA256: "schema-digest", + }) + if err != nil { + t.Fatal(err) + } + event := &adksession.Event{ + Author: "root", + LLMResponse: model.LLMResponse{ + Content: genai.NewContentFromText(`{"answer":4}`, genai.RoleModel), + FinishReason: genai.FinishReasonStop, + }, + } + update := a2atype.NewArtifactEvent(&a2asrv.ExecutorContext{TaskID: "task-1", ContextID: "context-1"}, a2atype.NewTextPart(`{"answer":4}`)) + err = transformStructuredOutput(output, "root", event, update) + if err != nil { + t.Fatalf("transformStructuredOutput() error = %v", err) + } + if len(update.Artifact.Parts) != 1 || update.Artifact.Parts[0].MediaType != "application/json" { + t.Fatalf("structured artifact = %#v", update.Artifact) + } + want := map[string]any{"answer": float64(4)} + if got := update.Artifact.Parts[0].Data(); !maps.Equal(got.(map[string]any), want) { + t.Fatalf("structured data = %#v, want %#v", got, want) + } + if got, ok := apia2a.StructuredOutputSchemaSHA256(update.Artifact.Parts[0]); !ok || got != "schema-digest" { + t.Fatalf("schema digest = %#v", got) + } +} + +func TestNewKAgentExecutorRejectsInvalidOutputSchema(t *testing.T) { + executor, err := NewKAgentExecutor(KAgentExecutorConfig{ + Logger: slog.New(slog.DiscardHandler), + Output: &apiadk.OutputConfig{JSONSchema: []byte(`{`)}, + }) + if err == nil || executor != nil { + t.Fatalf("NewKAgentExecutor() = %#v, %v; want a construction error", executor, err) + } +} + +func TestNewKAgentExecutorRequiresRootAgent(t *testing.T) { + executor, err := NewKAgentExecutor(KAgentExecutorConfig{Logger: slog.New(slog.DiscardHandler)}) + if err == nil || executor != nil { + t.Fatalf("NewKAgentExecutor() = %#v, %v; want a construction error", executor, err) + } +} + +func TestStructuredOutputPartConverterDropsOnlyRootPartials(t *testing.T) { + converter := structuredOutputPartConverter(&structuredOutput{}, "root") + partial := &adksession.Event{ + Author: "root", + LLMResponse: model.LLMResponse{Partial: true}, + } + got, err := converter(context.Background(), partial, genai.NewPartFromText("partial JSON")) + if err != nil || got != nil { + t.Fatalf("root partial conversion = %#v, error %v; want nil", got, err) + } + + partial.Author = "tool-agent" + got, err = converter(context.Background(), partial, genai.NewPartFromText("tool progress")) + if err != nil || got == nil || got.Text() != "tool progress" { + t.Fatalf("non-root partial conversion = %#v, error %v", got, err) + } +} + +func TestTransformStructuredOutputRejectsInvalidValueWithoutLeakingIt(t *testing.T) { + output, err := resolveStructuredOutput(&apiadk.OutputConfig{ + JSONSchema: []byte(`{"type":"object","properties":{"secret":{"type":"integer"}},"required":["secret"]}`), + }) + if err != nil { + t.Fatal(err) + } + event := &adksession.Event{ + Author: "root", + LLMResponse: model.LLMResponse{ + Content: genai.NewContentFromText(`{"secret":"do-not-log"}`, genai.RoleModel), + FinishReason: genai.FinishReasonStop, + }, + } + update := a2atype.NewArtifactEvent(&a2asrv.ExecutorContext{TaskID: "task-1", ContextID: "context-1"}) + err = transformStructuredOutput(output, "root", event, update) + if err == nil || strings.Contains(err.Error(), "do-not-log") { + t.Fatalf("validation error = %v", err) + } +} + +func TestTransformStructuredOutputRejectsIncompleteResponse(t *testing.T) { + output, err := resolveStructuredOutput(&apiadk.OutputConfig{ + JSONSchema: []byte(`{"type":"object","properties":{"answer":{"type":"integer"}},"required":["answer"]}`), + }) + if err != nil { + t.Fatal(err) + } + event := &adksession.Event{ + Author: "root", + LLMResponse: model.LLMResponse{ + Content: genai.NewContentFromText(`{"answer":4}`, genai.RoleModel), + FinishReason: genai.FinishReasonMaxTokens, + }, + } + update := a2atype.NewArtifactEvent(&a2asrv.ExecutorContext{TaskID: "task-1", ContextID: "context-1"}, a2atype.NewTextPart(`{"answer":4}`)) + err = transformStructuredOutput(output, "root", event, update) + if err == nil || !strings.Contains(err.Error(), "did not complete") { + t.Fatalf("transformStructuredOutput() error = %v, want incomplete-output error", err) + } +} + +func TestTransformStructuredOutputAcceptsNormalFinishReasons(t *testing.T) { + output, err := resolveStructuredOutput(&apiadk.OutputConfig{ + JSONSchema: []byte(`{"type":"object","properties":{"answer":{"type":"integer"}},"required":["answer"]}`), + }) + if err != nil { + t.Fatal(err) + } + for _, test := range []struct { + name string + reason genai.FinishReason + }{ + {name: "empty", reason: ""}, + {name: "unspecified", reason: genai.FinishReasonUnspecified}, + {name: "stop", reason: genai.FinishReasonStop}, + } { + t.Run(test.name, func(t *testing.T) { + event := &adksession.Event{ + Author: "root", + LLMResponse: model.LLMResponse{ + Content: genai.NewContentFromText(`{"answer":4}`, genai.RoleModel), + FinishReason: test.reason, + }, + } + update := a2atype.NewArtifactEvent(&a2asrv.ExecutorContext{}, a2atype.NewTextPart(`{"answer":4}`)) + if err := transformStructuredOutput(output, "root", event, update); err != nil { + t.Fatalf("transformStructuredOutput() error = %v", err) + } + if len(update.Artifact.Parts) != 1 || !apia2a.IsStructuredOutputPart(update.Artifact.Parts[0]) { + t.Fatalf("artifact parts = %#v, want structured output", update.Artifact.Parts) + } + }) + } +} + +func TestTransformStructuredOutputIgnoresSkipSummarizationEvent(t *testing.T) { + output, err := resolveStructuredOutput(&apiadk.OutputConfig{JSONSchema: []byte(`{"type":"object"}`)}) + if err != nil { + t.Fatal(err) + } + event := &adksession.Event{ + Author: "root", + Actions: adksession.EventActions{SkipSummarization: true}, + LLMResponse: model.LLMResponse{ + Content: genai.NewContentFromText("not a structured result", genai.RoleModel), + }, + } + update := a2atype.NewArtifactEvent(&a2asrv.ExecutorContext{}, a2atype.NewTextPart("not a structured result")) + if err := transformStructuredOutput(output, "root", event, update); err != nil { + t.Fatalf("transformStructuredOutput() error = %v", err) + } + if update.Artifact.Parts[0].Text() != "not a structured result" { + t.Fatalf("artifact was transformed: %#v", update.Artifact.Parts) + } +} + +func TestKAgentExecutorStructuredOutputAllowsHITLPause(t *testing.T) { + tests := []struct { + name string + toolName string + toolArgs map[string]any + isAsk bool + }{ + { + name: "tool approval", + toolName: "delete_file", + toolArgs: map[string]any{"path": "/tmp/x"}, + }, + { + name: "ask user", + toolName: "ask_user", + toolArgs: map[string]any{"questions": []any{map[string]any{"question": "Which namespace?"}}}, + isAsk: true, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + agent, err := adkagent.New(adkagent.Config{ + Name: "structured-agent", + Run: func(ic adkagent.InvocationContext) iter.Seq2[*adksession.Event, error] { + return func(yield func(*adksession.Event, error) bool) { + call := genai.NewPartFromFunctionCall(toolconfirmation.FunctionCallName, map[string]any{ + "originalFunctionCall": map[string]any{ + "name": test.toolName, + "id": "tool-call", + "args": test.toolArgs, + }, + "toolConfirmation": map[string]any{ + "hint": "Human input required", + "confirmed": false, + }, + }) + call.FunctionCall.ID = "confirmation-call" + yield(&adksession.Event{ + Author: ic.Agent().Name(), + InvocationID: ic.InvocationID(), + LLMResponse: model.LLMResponse{Content: &genai.Content{Role: genai.RoleModel, Parts: []*genai.Part{call}}}, + LongRunningToolIDs: []string{"confirmation-call"}, + }, nil) + } + }, + }) + if err != nil { + t.Fatal(err) + } + executor, err := NewKAgentExecutor(KAgentExecutorConfig{ + AppName: "test-app", + SessionService: adksession.InMemoryService(), + Logger: slog.New(slog.DiscardHandler), + RunnerConfig: runner.Config{AppName: "test-app", Agent: agent}, + Output: &apiadk.OutputConfig{ + JSONSchema: []byte(`{"type":"object","properties":{"answer":{"type":"string"}},"required":["answer"]}`), + SHA256: "schema-digest", + }, + }) + if err != nil { + t.Fatal(err) + } + + ctx, callCtx := a2asrv.NewCallContext(t.Context(), a2asrv.NewServiceParams(map[string][]string{ + a2atype.SvcParamExtensions: {HITLExtensionURI}, + })) + if _, _, err := HITLActivationInterceptor().Before(ctx, callCtx, &a2asrv.Request{}); err != nil { + t.Fatal(err) + } + reqCtx := &a2asrv.ExecutorContext{ + TaskID: "task-1", + ContextID: "context-1", + Message: a2atype.NewMessage(a2atype.MessageRoleUser, a2atype.NewTextPart("continue")), + } + + var pause *a2atype.TaskStatusUpdateEvent + for event, err := range executor.Execute(ctx, reqCtx) { + if err != nil { + t.Fatalf("Execute() error = %v", err) + } + if update, ok := event.(*a2atype.TaskStatusUpdateEvent); ok { + if update.Status.State == a2atype.TaskStateFailed { + t.Fatalf("structured HITL pause failed: %#v", update.Status.Message) + } + if update.Status.State == a2atype.TaskStateInputRequired { + pause = update + } + } + } + if pause == nil || pause.Status.Message == nil { + t.Fatalf("pause = %#v, want input-required status", pause) + } + if test.isAsk { + if request := GetAskUserRequest(pause.Status.Message); request == nil || len(request.Questions) != 1 { + t.Fatalf("ask-user request = %#v", request) + } + } else if request := GetToolApprovalRequest(pause.Status.Message); request == nil || len(request.Tools) != 1 { + t.Fatalf("tool-approval request = %#v", request) + } + }) + } +} + +func TestKAgentExecutorFailsCompletedStructuredOutputWithoutResult(t *testing.T) { + agent, err := adkagent.New(adkagent.Config{ + Name: "structured-agent", + Run: func(ic adkagent.InvocationContext) iter.Seq2[*adksession.Event, error] { + return func(yield func(*adksession.Event, error) bool) { + yield(&adksession.Event{ + Author: ic.Agent().Name(), + InvocationID: ic.InvocationID(), + LLMResponse: model.LLMResponse{ + Content: &genai.Content{Role: genai.RoleModel}, + FinishReason: genai.FinishReasonStop, + }, + }, nil) + } + }, + }) + if err != nil { + t.Fatal(err) + } + sessionService := adksession.InMemoryService() + executor, err := NewKAgentExecutor(KAgentExecutorConfig{ + AppName: "test-app", + SessionService: sessionService, + Logger: slog.New(slog.DiscardHandler), + RunnerConfig: runner.Config{AppName: "test-app", Agent: agent}, + Output: &apiadk.OutputConfig{ + JSONSchema: []byte(`{"type":"object"}`), + SHA256: "schema-digest", + }, + }) + if err != nil { + t.Fatal(err) + } + reqCtx := &a2asrv.ExecutorContext{ + TaskID: "task-1", ContextID: "context-1", + Message: a2atype.NewMessage(a2atype.MessageRoleUser, a2atype.NewTextPart("answer")), + } + + var terminal *a2atype.TaskStatusUpdateEvent + for event, err := range executor.Execute(t.Context(), reqCtx) { + if err != nil { + t.Fatal(err) + } + if update, ok := event.(*a2atype.TaskStatusUpdateEvent); ok && update.Status.State.Terminal() { + terminal = update + } + } + if terminal == nil || terminal.Status.State != a2atype.TaskStateFailed || terminal.Status.Message == nil { + t.Fatalf("terminal status = %#v, want failed status with an error message", terminal) + } +} + func TestKAgentExecutor_StreamsArtifactsThroughUpstreamExecutor(t *testing.T) { const ( appName = "test-app" @@ -244,7 +568,7 @@ func TestKAgentExecutor_StreamsArtifactsThroughUpstreamExecutor(t *testing.T) { } sessionService := adksession.InMemoryService() - executor := NewKAgentExecutor(KAgentExecutorConfig{ + executor, err := NewKAgentExecutor(KAgentExecutorConfig{ AppName: appName, SessionService: sessionService, Logger: slog.New(slog.DiscardHandler), @@ -253,6 +577,9 @@ func TestKAgentExecutor_StreamsArtifactsThroughUpstreamExecutor(t *testing.T) { Agent: agent, }, }) + if err != nil { + t.Fatalf("NewKAgentExecutor() error = %v", err) + } reqCtx := &a2asrv.ExecutorContext{ TaskID: "task-1", ContextID: contextID, @@ -320,7 +647,10 @@ func TestKAgentExecutor_HITLPauseAndResumeFlow(t *testing.T) { } yield(&adksession.Event{ Author: ic.Agent().Name(), InvocationID: ic.InvocationID(), Branch: ic.Branch(), - LLMResponse: model.LLMResponse{Content: genai.NewContentFromText("resumed", genai.RoleModel)}, + LLMResponse: model.LLMResponse{ + Content: genai.NewContentFromText(`{"answer":"resumed"}`, genai.RoleModel), + FinishReason: genai.FinishReasonStop, + }, }, nil) } }, @@ -329,10 +659,17 @@ func TestKAgentExecutor_HITLPauseAndResumeFlow(t *testing.T) { t.Fatalf("agent.New() error = %v", err) } sessionService := adksession.InMemoryService() - executor := NewKAgentExecutor(KAgentExecutorConfig{ + executor, err := NewKAgentExecutor(KAgentExecutorConfig{ AppName: appName, SessionService: sessionService, Logger: slog.New(slog.DiscardHandler), RunnerConfig: runner.Config{AppName: appName, Agent: agent}, + Output: &apiadk.OutputConfig{ + JSONSchema: []byte(`{"type":"object","properties":{"answer":{"type":"string"}},"required":["answer"]}`), + SHA256: "schema-digest", + }, }) + if err != nil { + t.Fatalf("NewKAgentExecutor() error = %v", err) + } ctx, callCtx := a2asrv.NewCallContext(context.Background(), a2asrv.NewServiceParams(map[string][]string{ a2atype.SvcParamExtensions: {HITLExtensionURI}, })) @@ -353,8 +690,11 @@ func TestKAgentExecutor_HITLPauseAndResumeFlow(t *testing.T) { pause = update } } + if pause == nil { + t.Fatalf("pause = %#v, want extension input-required", pause) + } req := GetToolApprovalRequest(pause.Status.Message) - if pause == nil || req == nil { + if req == nil { t.Fatalf("pause = %#v, want extension input-required", pause) } if _, ok := pause.Status.Message.Metadata[apia2a.TimelinePositionMetadataKey].(string); !ok { @@ -375,17 +715,25 @@ func TestKAgentExecutor_HITLPauseAndResumeFlow(t *testing.T) { resume := &a2asrv.ExecutorContext{ TaskID: "hitl-task", ContextID: contextID, Message: decision, StoredTask: stored, } - var resumedText string + var resumedJSON string for event, err := range executor.Execute(ctx, resume) { if err != nil { t.Fatalf("resume Execute() error = %v", err) } if artifact, ok := event.(*a2atype.TaskArtifactUpdateEvent); ok && len(artifact.Artifact.Parts) > 0 { - resumedText = artifact.Artifact.Parts[0].Text() + for _, part := range artifact.Artifact.Parts { + if !apia2a.IsStructuredOutputPart(part) { + continue + } + resumedJSON, err = apia2a.StructuredOutputJSON(part) + if err != nil { + t.Fatal(err) + } + } } } - if resumedText != "resumed" || invocations != 2 { - t.Fatalf("resumed text = %q, invocations = %d", resumedText, invocations) + if resumedJSON != `{"answer":"resumed"}` || invocations != 2 { + t.Fatalf("resumed output = %q, invocations = %d", resumedJSON, invocations) } } diff --git a/go/adk/pkg/a2a/fork_continuity_test.go b/go/adk/pkg/a2a/fork_continuity_test.go index 2efd9efead..32d2f85aac 100644 --- a/go/adk/pkg/a2a/fork_continuity_test.go +++ b/go/adk/pkg/a2a/fork_continuity_test.go @@ -46,10 +46,12 @@ func TestKAgentExecutorForkRetainsConversation(t *testing.T) { t.Helper() svc, err := localsession.NewLocalSessionService("sqlite:///" + path) require.NoError(t, err) - return NewKAgentExecutor(KAgentExecutorConfig{ + executor, err := NewKAgentExecutor(KAgentExecutorConfig{ AppName: "app", SessionService: svc, Logger: slog.New(slog.DiscardHandler), RunnerConfig: runner.Config{AppName: "app", Agent: agent}, }) + require.NoError(t, err) + return executor } send := func(executor *KAgentExecutor, contextID string, eventCount int) { t.Helper() diff --git a/go/adk/pkg/agent/agent.go b/go/adk/pkg/agent/agent.go index d8b1f6a524..0917ad5bb0 100644 --- a/go/adk/pkg/agent/agent.go +++ b/go/adk/pkg/agent/agent.go @@ -2,6 +2,7 @@ package agent import ( "context" + "encoding/json" "fmt" "os" "strings" @@ -10,6 +11,7 @@ import ( "github.com/kagent-dev/kagent/go/adk/pkg/mcp" "github.com/kagent-dev/kagent/go/adk/pkg/models" + adkoutputschema "github.com/kagent-dev/kagent/go/adk/pkg/outputschema" "github.com/kagent-dev/kagent/go/adk/pkg/sts" "github.com/kagent-dev/kagent/go/adk/pkg/tools" "github.com/kagent-dev/kagent/go/api/adk" @@ -154,6 +156,25 @@ func createGoogleADKAgent(ctx context.Context, agentConfig *adk.AgentConfig, age makeOnToolErrorCallback(log), }, } + if agentConfig.Output != nil { + llmAgentConfig.OutputSchema, err = adkoutputschema.ToGenAISchema(agentConfig.Output.JSONSchema) + if err != nil { + return nil, err + } + // Provider adapters with native raw-schema fields consume the canonical + // JSON Schema directly.Gemini must use only ADK's OutputSchema projection. + if usesRawOutputSchema(agentConfig.Model) { + if llmAgentConfig.GenerateContentConfig == nil { + llmAgentConfig.GenerateContentConfig = &genai.GenerateContentConfig{} + } + var rawSchema any + if err := json.Unmarshal(agentConfig.Output.JSONSchema, &rawSchema); err != nil { + return nil, fmt.Errorf("decode raw output schema: %w", err) + } + llmAgentConfig.GenerateContentConfig.ResponseJsonSchema = rawSchema + llmAgentConfig.GenerateContentConfig.ResponseMIMEType = "application/json" + } + } log.InfoContext(ctx, "creating Google ADK LLM agent", "name", llmAgentConfig.Name, @@ -174,6 +195,18 @@ func createGoogleADKAgent(ctx context.Context, agentConfig *adk.AgentConfig, age return llmAgent, nil } +// usesRawOutputSchema reports whether the kagent provider adapter has a native +// raw JSON Schema request field. Gemini is intentionally absent: upstream ADK +// owns its native-vs-set_model_response selection using the typed projection. +func usesRawOutputSchema(model adk.Model) bool { + switch model.(type) { + case *adk.OpenAI, *adk.AzureOpenAI, *adk.Anthropic, *adk.GeminiAnthropic, *adk.Bedrock, *adk.Foundry: + return true + default: + return false + } +} + func buildAgentTools(agentConfig *adk.AgentConfig, remoteAgentTools, extraTools []tool.Tool, log *slog.Logger) ([]tool.Tool, error) { var localTools []tool.Tool if agentConfig.Memory != nil { diff --git a/go/adk/pkg/agent/agent_test.go b/go/adk/pkg/agent/agent_test.go index 768b086737..893d74ac88 100644 --- a/go/adk/pkg/agent/agent_test.go +++ b/go/adk/pkg/agent/agent_test.go @@ -7,6 +7,7 @@ import ( "github.com/kagent-dev/kagent/go/adk/pkg/models" "github.com/kagent-dev/kagent/go/api/adk" + "github.com/stretchr/testify/require" ) // TestConfigDeserialization_OpenAI verifies that a realistic OpenAI config.json @@ -463,3 +464,26 @@ func TestCreateGoogleADKAgentBuildsSubAgents(t *testing.T) { t.Fatalf("sub-agents = %#v", root.SubAgents()) } } + +func TestUsesRawOutputSchema(t *testing.T) { + tests := []struct { + name string + model adk.Model + want bool + }{ + {name: "OpenAI", model: &adk.OpenAI{}, want: true}, + {name: "Azure OpenAI", model: &adk.AzureOpenAI{}, want: true}, + {name: "Anthropic", model: &adk.Anthropic{}, want: true}, + {name: "Anthropic on Vertex", model: &adk.GeminiAnthropic{}, want: true}, + {name: "Bedrock Converse", model: &adk.Bedrock{}, want: true}, + {name: "Foundry OpenAI", model: &adk.Foundry{}, want: true}, + {name: "Foundry Anthropic", model: &adk.Foundry{APIFormat: adk.FoundryAPIFormatAnthropic}, want: true}, + {name: "Gemini", model: &adk.Gemini{}, want: false}, + {name: "Ollama", model: &adk.Ollama{}, want: false}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + require.Equal(t, test.want, usesRawOutputSchema(test.model)) + }) + } +} diff --git a/go/adk/pkg/models/anthropic_adk.go b/go/adk/pkg/models/anthropic_adk.go index c51a7da98c..8546db8685 100644 --- a/go/adk/pkg/models/anthropic_adk.go +++ b/go/adk/pkg/models/anthropic_adk.go @@ -73,6 +73,12 @@ func (m *AnthropicModel) GenerateContent(ctx context.Context, req *model.LLMRequ // Apply config options applyAnthropicConfig(¶ms, m.Config) + if schema, err := structuredOutputSchema(req.Config); err != nil { + yield(nil, err) + return + } else if schema != nil { + params.OutputConfig.Format = anthropic.JSONOutputFormatParam{Schema: schema} + } // Add tools if provided if req.Config != nil && len(req.Config.Tools) > 0 { diff --git a/go/adk/pkg/models/anthropic_adk_test.go b/go/adk/pkg/models/anthropic_adk_test.go new file mode 100644 index 0000000000..b73f53e249 --- /dev/null +++ b/go/adk/pkg/models/anthropic_adk_test.go @@ -0,0 +1,72 @@ +package models + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "testing" + + "google.golang.org/adk/v2/model" + "google.golang.org/genai" +) + +func TestAnthropicModelGenerateContentSendsStructuredOutputWithTools(t *testing.T) { + var body map[string]any + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + encoded, _ := io.ReadAll(r.Body) + _ = json.Unmarshal(encoded, &body) + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, `{ + "id":"msg-1","type":"message","role":"assistant","model":"claude-sonnet-4-5", + "content":[{"type":"text","text":"{\"answer\":4}"}], + "stop_reason":"end_turn","stop_sequence":null, + "usage":{"input_tokens":1,"output_tokens":1} + }`) + })) + defer server.Close() + + llm, err := newAnthropicModelFromConfig(context.Background(), &AnthropicConfig{ + Model: "claude-sonnet-4-5", + BaseUrl: server.URL, + }, "test-key") + if err != nil { + t.Fatal(err) + } + schema := map[string]any{ + "type": "object", + "properties": map[string]any{"answer": map[string]any{"type": "integer"}}, + "required": []any{"answer"}, + "additionalProperties": false, + } + request := &model.LLMRequest{ + Contents: []*genai.Content{{Role: "user", Parts: []*genai.Part{{Text: "calculate"}}}}, + Config: &genai.GenerateContentConfig{ + ResponseJsonSchema: schema, + Tools: []*genai.Tool{{FunctionDeclarations: []*genai.FunctionDeclaration{{ + Name: "calculator", ParametersJsonSchema: map[string]any{"type": "object", "properties": map[string]any{}}, + }}}}, + }, + } + for _, generateErr := range llm.GenerateContent(context.Background(), request, false) { + if generateErr != nil { + t.Fatalf("GenerateContent error: %v", generateErr) + } + } + + outputConfig, ok := body["output_config"].(map[string]any) + if !ok { + t.Fatalf("output_config = %#v", body["output_config"]) + } + format, ok := outputConfig["format"].(map[string]any) + if !ok || format["type"] != "json_schema" { + t.Fatalf("output_config.format = %#v", outputConfig["format"]) + } + if gotSchema, ok := format["schema"].(map[string]any); !ok || gotSchema["additionalProperties"] != false { + t.Fatalf("output_config.format.schema = %#v", format["schema"]) + } + if tools, ok := body["tools"].([]any); !ok || len(tools) != 1 { + t.Fatalf("tools = %#v, want one tool", body["tools"]) + } +} diff --git a/go/adk/pkg/models/bedrock.go b/go/adk/pkg/models/bedrock.go index e7d8701753..05aa74a01d 100644 --- a/go/adk/pkg/models/bedrock.go +++ b/go/adk/pkg/models/bedrock.go @@ -247,18 +247,47 @@ func (m *BedrockModel) GenerateContent(ctx context.Context, req *model.LLMReques } additionalFields := m.buildAdditionalModelRequestFields() + outputConfig, err := bedrockOutputConfig(req.Config) + if err != nil { + yield(nil, err) + return + } // Set telemetry attributes telemetry.SetLLMRequestAttributes(ctx, modelName, req) if stream { - m.generateStreaming(ctx, modelName, messages, systemPrompt, inferenceConfig, toolConfig, additionalFields, reverseNameMap, yield) + m.generateStreaming(ctx, modelName, messages, systemPrompt, inferenceConfig, toolConfig, outputConfig, additionalFields, reverseNameMap, yield) } else { - m.generateNonStreaming(ctx, modelName, messages, systemPrompt, inferenceConfig, toolConfig, additionalFields, reverseNameMap, yield) + m.generateNonStreaming(ctx, modelName, messages, systemPrompt, inferenceConfig, toolConfig, outputConfig, additionalFields, reverseNameMap, yield) } } } +// bedrockOutputConfig maps ADK's response schema to the Bedrock Converse +// structured-output shape. +func bedrockOutputConfig(config *genai.GenerateContentConfig) (*types.OutputConfig, error) { + schema, err := structuredOutputSchema(config) + if err != nil || schema == nil { + return nil, err + } + encoded, err := json.Marshal(schema) + if err != nil { + return nil, fmt.Errorf("marshal Bedrock output schema: %w", err) + } + return &types.OutputConfig{ + TextFormat: &types.OutputFormat{ + Type: types.OutputFormatTypeJsonSchema, + Structure: &types.OutputFormatStructureMemberJsonSchema{ + Value: types.JsonSchemaDefinition{ + Name: aws.String(structuredOutputName), + Schema: aws.String(string(encoded)), + }, + }, + }, + }, nil +} + // buildAdditionalModelRequestFields returns a document.Interface containing // model-specific parameters that are not part of InferenceConfiguration. // The raw map is forwarded as-is to the Bedrock Converse API. @@ -273,13 +302,14 @@ func (m *BedrockModel) buildAdditionalModelRequestFields() document.Interface { // generateStreaming handles streaming responses from Bedrock ConverseStream. // It properly handles both text and tool use content blocks during streaming. // reverseNameMap maps sanitized Bedrock tool names back to their original names. -func (m *BedrockModel) generateStreaming(ctx context.Context, modelId string, messages []types.Message, systemPrompt []types.SystemContentBlock, inferenceConfig *types.InferenceConfiguration, toolConfig *types.ToolConfiguration, additionalFields document.Interface, reverseNameMap map[string]string, yield func(*model.LLMResponse, error) bool) { +func (m *BedrockModel) generateStreaming(ctx context.Context, modelId string, messages []types.Message, systemPrompt []types.SystemContentBlock, inferenceConfig *types.InferenceConfiguration, toolConfig *types.ToolConfiguration, outputConfig *types.OutputConfig, additionalFields document.Interface, reverseNameMap map[string]string, yield func(*model.LLMResponse, error) bool) { output, err := m.Client.ConverseStream(ctx, &bedrockruntime.ConverseStreamInput{ ModelId: aws.String(modelId), Messages: messages, System: systemPrompt, InferenceConfig: inferenceConfig, ToolConfig: toolConfig, + OutputConfig: outputConfig, AdditionalModelRequestFields: additionalFields, GuardrailConfig: bedrockGuardrailStreamConfig(m.Config), }) @@ -467,13 +497,14 @@ func (tc *streamingToolCall) parseArgs() map[string]any { // generateNonStreaming handles non-streaming responses from Bedrock Converse. // reverseNameMap maps sanitized Bedrock tool names back to their original names. -func (m *BedrockModel) generateNonStreaming(ctx context.Context, modelId string, messages []types.Message, systemPrompt []types.SystemContentBlock, inferenceConfig *types.InferenceConfiguration, toolConfig *types.ToolConfiguration, additionalFields document.Interface, reverseNameMap map[string]string, yield func(*model.LLMResponse, error) bool) { +func (m *BedrockModel) generateNonStreaming(ctx context.Context, modelId string, messages []types.Message, systemPrompt []types.SystemContentBlock, inferenceConfig *types.InferenceConfiguration, toolConfig *types.ToolConfiguration, outputConfig *types.OutputConfig, additionalFields document.Interface, reverseNameMap map[string]string, yield func(*model.LLMResponse, error) bool) { output, err := m.Client.Converse(ctx, &bedrockruntime.ConverseInput{ ModelId: aws.String(modelId), Messages: messages, System: systemPrompt, InferenceConfig: inferenceConfig, ToolConfig: toolConfig, + OutputConfig: outputConfig, AdditionalModelRequestFields: additionalFields, GuardrailConfig: bedrockGuardrailConfig(m.Config), }) diff --git a/go/adk/pkg/models/bedrock_test.go b/go/adk/pkg/models/bedrock_test.go index a7a262021d..e253029de1 100644 --- a/go/adk/pkg/models/bedrock_test.go +++ b/go/adk/pkg/models/bedrock_test.go @@ -1,17 +1,103 @@ package models import ( + "context" "encoding/json" + "io" + "net/http" + "net/http/httptest" "strings" "testing" + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/credentials" + "github.com/aws/aws-sdk-go-v2/service/bedrockruntime" "github.com/aws/aws-sdk-go-v2/service/bedrockruntime/types" "github.com/google/jsonschema-go/jsonschema" + "google.golang.org/adk/v2/model" "google.golang.org/genai" ) // testLogger returns a no-op logger for tests. +func TestBedrockModelGenerateContentSendsStructuredOutputWithTools(t *testing.T) { + t.Setenv("AWS_BEARER_TOKEN_BEDROCK", "") + var body map[string]any + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + encoded, _ := io.ReadAll(r.Body) + _ = json.Unmarshal(encoded, &body) + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, `{ + "output":{"message":{"role":"assistant","content":[{"text":"{\"answer\":4}"}]}}, + "stopReason":"end_turn", + "usage":{"inputTokens":1,"outputTokens":1,"totalTokens":2}, + "metrics":{"latencyMs":1} + }`) + })) + defer server.Close() + + client := bedrockruntime.New(bedrockruntime.Options{ + Region: "us-east-1", + BaseEndpoint: aws.String(server.URL), + HTTPClient: server.Client(), + Credentials: credentials.NewStaticCredentialsProvider("test", "test", ""), + RetryMaxAttempts: 1, + }) + llm := &BedrockModel{Config: &BedrockConfig{Model: "anthropic.claude-sonnet-4-5-v1:0"}, Client: client} + schema := map[string]any{ + "type": "object", + "properties": map[string]any{"answer": map[string]any{"type": "integer"}}, + "required": []any{"answer"}, + "additionalProperties": false, + } + request := &model.LLMRequest{ + Contents: []*genai.Content{{Role: "user", Parts: []*genai.Part{{Text: "calculate"}}}}, + Config: &genai.GenerateContentConfig{ + ResponseJsonSchema: schema, + Tools: []*genai.Tool{{FunctionDeclarations: []*genai.FunctionDeclaration{{ + Name: "calculator", ParametersJsonSchema: map[string]any{"type": "object", "properties": map[string]any{}}, + }}}}, + }, + } + for _, generateErr := range llm.GenerateContent(context.Background(), request, false) { + if generateErr != nil { + t.Fatalf("GenerateContent error: %v", generateErr) + } + } + + outputConfig, ok := body["outputConfig"].(map[string]any) + if !ok { + t.Fatalf("outputConfig = %#v", body["outputConfig"]) + } + textFormat, ok := outputConfig["textFormat"].(map[string]any) + if !ok || textFormat["type"] != "json_schema" { + t.Fatalf("outputConfig.textFormat = %#v", outputConfig["textFormat"]) + } + structure, ok := textFormat["structure"].(map[string]any) + if !ok { + t.Fatalf("outputConfig.textFormat.structure = %#v", textFormat["structure"]) + } + jsonSchema, ok := structure["jsonSchema"].(map[string]any) + if !ok || jsonSchema["name"] != "kagent_output" { + t.Fatalf("outputConfig.textFormat.structure.jsonSchema = %#v", structure["jsonSchema"]) + } + encodedSchema, ok := jsonSchema["schema"].(string) + if !ok { + t.Fatalf("outputConfig JSON schema = %#v, want string", jsonSchema["schema"]) + } + var gotSchema map[string]any + if err := json.Unmarshal([]byte(encodedSchema), &gotSchema); err != nil || gotSchema["additionalProperties"] != false { + t.Fatalf("Bedrock output schema = %#v, error %v", jsonSchema["schema"], err) + } + toolConfig, ok := body["toolConfig"].(map[string]any) + if !ok { + t.Fatalf("toolConfig = %#v", body["toolConfig"]) + } + if tools, ok := toolConfig["tools"].([]any); !ok || len(tools) != 1 { + t.Fatalf("toolConfig.tools = %#v, want one tool", toolConfig["tools"]) + } +} + func TestBedrockStopReasonToGenai(t *testing.T) { tests := []struct { name string diff --git a/go/adk/pkg/models/openai_adk.go b/go/adk/pkg/models/openai_adk.go index 89665545f1..d4ecefa097 100644 --- a/go/adk/pkg/models/openai_adk.go +++ b/go/adk/pkg/models/openai_adk.go @@ -172,6 +172,16 @@ func generateContentChatCompletions( }, params.Messages...) } applyOpenAIConfig(¶ms, m.Config) + if schema, err := structuredOutputSchema(req.Config); err != nil { + yield(nil, err) + return + } else if schema != nil { + params.ResponseFormat = openai.ChatCompletionNewParamsResponseFormatUnion{ + OfJSONSchema: &shared.ResponseFormatJSONSchemaParam{JSONSchema: shared.ResponseFormatJSONSchemaJSONSchemaParam{ + Name: structuredOutputName, Schema: schema, + }}, + } + } if req.Config != nil && len(req.Config.Tools) > 0 { params.Tools = genaiToolsToOpenAITools(req.Config.Tools) diff --git a/go/adk/pkg/models/openai_adk_test.go b/go/adk/pkg/models/openai_adk_test.go index dde5959be8..c5ba95d671 100644 --- a/go/adk/pkg/models/openai_adk_test.go +++ b/go/adk/pkg/models/openai_adk_test.go @@ -1,11 +1,19 @@ package models import ( + "context" "encoding/base64" "encoding/json" + "io" + "net/http" + "net/http/httptest" "testing" + "log/slog" + "github.com/openai/openai-go/v3" + "github.com/openai/openai-go/v3/option" + "google.golang.org/adk/v2/model" "google.golang.org/genai" ) @@ -16,6 +24,67 @@ func TestOpenAIModel_Name(t *testing.T) { } } +func TestOpenAIModelGenerateContentSendsStructuredOutputWithTools(t *testing.T) { + var body map[string]any + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + encoded, _ := io.ReadAll(r.Body) + _ = json.Unmarshal(encoded, &body) + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, `{ + "id":"chatcmpl-1","object":"chat.completion","created":1,"model":"gpt-4o", + "choices":[{"index":0,"message":{"role":"assistant","content":"{\"answer\":4}"},"finish_reason":"stop"}] + }`) + })) + defer server.Close() + + client := openai.NewClient( + option.WithAPIKey("test"), + option.WithBaseURL(server.URL), + option.WithHTTPClient(server.Client()), + ) + llm := &OpenAIModel{ + Config: &OpenAIConfig{Model: "gpt-4o"}, + Client: client, + Logger: slog.New(slog.DiscardHandler), + } + schema := map[string]any{ + "type": "object", + "properties": map[string]any{"answer": map[string]any{"type": "integer"}}, + } + request := &model.LLMRequest{ + Contents: []*genai.Content{{Role: "user", Parts: []*genai.Part{{Text: "calculate"}}}}, + Config: &genai.GenerateContentConfig{ + ResponseJsonSchema: schema, + Tools: []*genai.Tool{{FunctionDeclarations: []*genai.FunctionDeclaration{{ + Name: "calculator", ParametersJsonSchema: map[string]any{"type": "object", "properties": map[string]any{}}, + }}}}, + }, + } + for _, err := range llm.GenerateContent(context.Background(), request, false) { + if err != nil { + t.Fatalf("GenerateContent error: %v", err) + } + } + + responseFormat, ok := body["response_format"].(map[string]any) + if !ok || responseFormat["type"] != "json_schema" { + t.Fatalf("response_format = %#v", body["response_format"]) + } + jsonSchema, ok := responseFormat["json_schema"].(map[string]any) + if !ok { + t.Fatalf("response_format.json_schema = %#v", responseFormat["json_schema"]) + } + if _, present := jsonSchema["strict"]; present { + t.Fatalf("response_format.json_schema.strict must be omitted: %#v", jsonSchema) + } + if gotSchema, ok := jsonSchema["schema"].(map[string]any); !ok || gotSchema["type"] != "object" { + t.Fatalf("response_format.json_schema.schema = %#v", jsonSchema["schema"]) + } + if tools, ok := body["tools"].([]any); !ok || len(tools) != 1 { + t.Fatalf("tools = %#v, want one tool", body["tools"]) + } +} + func TestFunctionResponseContentString(t *testing.T) { tests := []struct { name string diff --git a/go/adk/pkg/models/openai_responses.go b/go/adk/pkg/models/openai_responses.go index a39810dd58..ab21f6af75 100644 --- a/go/adk/pkg/models/openai_responses.go +++ b/go/adk/pkg/models/openai_responses.go @@ -36,6 +36,13 @@ func generateContentResponses( params.Instructions = param.NewOpt(instructions) } applyOpenAIResponsesConfig(¶ms, m.Config) + if schema, err := structuredOutputSchema(req.Config); err != nil { + yield(nil, err) + return + } else if schema != nil { + format := responses.ResponseFormatTextConfigParamOfJSONSchema(structuredOutputName, schema) + params.Text.Format = format + } if req.Config != nil && len(req.Config.Tools) > 0 { params.Tools = genaiToolsToResponsesTools(req.Config.Tools) diff --git a/go/adk/pkg/models/openai_responses_test.go b/go/adk/pkg/models/openai_responses_test.go index bbc9e1b0b4..cbec6bd103 100644 --- a/go/adk/pkg/models/openai_responses_test.go +++ b/go/adk/pkg/models/openai_responses_test.go @@ -178,8 +178,18 @@ func TestOpenAIModel_GenerateContent_Responses(t *testing.T) { } var got *model.LLMResponse + schema := map[string]any{ + "type": "object", + "properties": map[string]any{"answer": map[string]any{"type": "integer"}}, + } for resp, err := range m.GenerateContent(context.Background(), &model.LLMRequest{ Contents: []*genai.Content{{Role: "user", Parts: []*genai.Part{{Text: "ping"}}}}, + Config: &genai.GenerateContentConfig{ + ResponseJsonSchema: schema, + Tools: []*genai.Tool{{FunctionDeclarations: []*genai.FunctionDeclaration{{ + Name: "calculator", ParametersJsonSchema: map[string]any{"type": "object", "properties": map[string]any{}}, + }}}}, + }, }, false) { if err != nil { t.Fatalf("GenerateContent error: %v", err) @@ -192,6 +202,23 @@ func TestOpenAIModel_GenerateContent_Responses(t *testing.T) { if gotBody["model"] != "gpt-4o" { t.Fatalf("body model = %#v", gotBody["model"]) } + textConfig, ok := gotBody["text"].(map[string]any) + if !ok { + t.Fatalf("body text = %#v", gotBody["text"]) + } + format, ok := textConfig["format"].(map[string]any) + if !ok || format["type"] != "json_schema" { + t.Fatalf("body text.format = %#v", textConfig["format"]) + } + if _, present := format["strict"]; present { + t.Fatalf("body text.format.strict must be omitted: %#v", format) + } + if gotSchema, ok := format["schema"].(map[string]any); !ok || gotSchema["type"] != "object" { + t.Fatalf("body text.format.schema = %#v", format["schema"]) + } + if tools, ok := gotBody["tools"].([]any); !ok || len(tools) != 1 { + t.Fatalf("body tools = %#v, want one tool", gotBody["tools"]) + } if got == nil || got.Content == nil || len(got.Content.Parts) != 1 || got.Content.Parts[0].Text != "pong" { t.Fatalf("response = %#v", got) } diff --git a/go/adk/pkg/models/structured_output.go b/go/adk/pkg/models/structured_output.go new file mode 100644 index 0000000000..0c146b4e97 --- /dev/null +++ b/go/adk/pkg/models/structured_output.go @@ -0,0 +1,66 @@ +package models + +import ( + "encoding/json" + "fmt" + "strings" + + "google.golang.org/genai" +) + +// structuredOutputName identifies kagent's response format to providers. It is +// an API request label, not a field in the generated result. +const structuredOutputName = "kagent_output" + +// structuredOutputSchema returns the JSON Schema that a provider request +// should send. +// +// Kagent places the canonical schema in ResponseJsonSchema for adapters that +// support raw JSON Schema, so prefer that value: it preserves keywords such as +// $defs, const, and additionalProperties that genai.Schema cannot represent. +// The ResponseSchema fallback also makes this helper safe for callers using +// ADK's typed schema directly. +func structuredOutputSchema(config *genai.GenerateContentConfig) (map[string]any, error) { + if config == nil || (config.ResponseJsonSchema == nil && config.ResponseSchema == nil) { + return nil, nil + } + if config.ResponseJsonSchema != nil { + schema, ok := config.ResponseJsonSchema.(map[string]any) + if !ok { + return nil, fmt.Errorf("response JSON schema has type %T, want map[string]any", config.ResponseJsonSchema) + } + return schema, nil + } + + encoded, err := json.Marshal(config.ResponseSchema) + if err != nil { + return nil, fmt.Errorf("marshal response schema: %w", err) + } + var schema map[string]any + if err := json.Unmarshal(encoded, &schema); err != nil { + return nil, fmt.Errorf("decode response schema: %w", err) + } + normalizeSchemaTypes(schema) + return schema, nil +} + +// normalizeSchemaTypes converts genai.Type's JSON values (for example, +// "OBJECT") to JSON Schema spelling ("object"). It is used only after +// marshaling a typed genai.Schema; raw JSON schemas must not be normalized, +// because an object-valued const or enum may legitimately contain a field +// named "type" whose value is case-sensitive user data. +func normalizeSchemaTypes(value any) { + switch value := value.(type) { + case map[string]any: + if typeName, ok := value["type"].(string); ok { + value["type"] = strings.ToLower(typeName) + } + for _, child := range value { + normalizeSchemaTypes(child) + } + case []any: + for _, child := range value { + normalizeSchemaTypes(child) + } + } +} diff --git a/go/adk/pkg/models/structured_output_test.go b/go/adk/pkg/models/structured_output_test.go new file mode 100644 index 0000000000..63acf164a1 --- /dev/null +++ b/go/adk/pkg/models/structured_output_test.go @@ -0,0 +1,52 @@ +package models + +import ( + "testing" + + "google.golang.org/genai" +) + +func TestStructuredOutputSchemaPrefersRawJSONSchema(t *testing.T) { + largeInteger := ^uint64(0) + schema, err := structuredOutputSchema(&genai.GenerateContentConfig{ + ResponseSchema: &genai.Schema{Type: genai.TypeObject}, + ResponseJsonSchema: map[string]any{ + "type": "object", "additionalProperties": false, "x-large-integer": largeInteger, + "properties": map[string]any{"result": map[string]any{"const": map[string]any{"type": "MixedCase"}}}, + }, + }) + if err != nil { + t.Fatal(err) + } + if schema["type"] != "object" || schema["additionalProperties"] != false { + t.Fatalf("schema = %#v", schema) + } + if schema["x-large-integer"] != largeInteger { + t.Fatalf("raw schema value was coerced: %#v", schema["x-large-integer"]) + } + constant := schema["properties"].(map[string]any)["result"].(map[string]any)["const"].(map[string]any) + if constant["type"] != "MixedCase" { + t.Fatalf("const value was modified: %#v", constant) + } +} + +func TestStructuredOutputSchemaRejectsNonObjectRawSchema(t *testing.T) { + _, err := structuredOutputSchema(&genai.GenerateContentConfig{ResponseJsonSchema: "not an object"}) + if err == nil { + t.Fatal("expected an error") + } +} + +func TestStructuredOutputSchemaNormalizesGenAITypeNames(t *testing.T) { + schema, err := structuredOutputSchema(&genai.GenerateContentConfig{ResponseSchema: &genai.Schema{ + Type: genai.TypeObject, + Properties: map[string]*genai.Schema{"answer": {Type: genai.TypeInteger}}, + }}) + if err != nil { + t.Fatal(err) + } + property := schema["properties"].(map[string]any)["answer"].(map[string]any) + if schema["type"] != "object" || property["type"] != "integer" { + t.Fatalf("schema = %#v", schema) + } +} diff --git a/go/adk/pkg/outputschema/convert.go b/go/adk/pkg/outputschema/convert.go new file mode 100644 index 0000000000..3c05b5a3c4 --- /dev/null +++ b/go/adk/pkg/outputschema/convert.go @@ -0,0 +1,146 @@ +// Package outputschema converts kagent's portable output schema into the +// narrower schema representation used by Go ADK. +package outputschema + +import ( + "encoding/json" + "fmt" + "strings" + + "github.com/google/jsonschema-go/jsonschema" + "google.golang.org/genai" +) + +const ( + maxConversionDepth = 32 + maxConversionNodes = 1_000 +) + +// ToGenAISchema converts a canonical kagent output schema to the genai.Schema used +// by Go ADK and also provides the concrete Go ADK compatibility check: +// recursive references and schemas whose expanded form exceeds the configured +// safety limits are rejected here. +// +// The canonical JSON Schema remains authoritative for provider requests that +// accept raw JSON Schema and for final response validation. genai.Schema lacks +// const and additionalProperties, and represents enum values as strings, so +// those constraints may be enforced only by the provider and final validator. +func ToGenAISchema(raw json.RawMessage) (*genai.Schema, error) { + var schema jsonschema.Schema + if err := json.Unmarshal(raw, &schema); err != nil { + return nil, fmt.Errorf("decode output schema: %w", err) + } + if _, err := schema.Resolve(nil); err != nil { + return nil, fmt.Errorf("resolve output schema: %w", err) + } + converter := schemaConverter{root: &schema} + return converter.convert(&schema, nil, 1) +} + +type schemaConverter struct { + root *jsonschema.Schema + nodes int +} + +// convert recursively converts a jsonschema.Schema to a genai.Schema, expanding +// local $defs references and enforcing the configured safety limits. The stack +// tracks the current recursion path to detect cycles. +func (c *schemaConverter) convert(schema *jsonschema.Schema, stack map[*jsonschema.Schema]struct{}, depth int) (*genai.Schema, error) { + if schema == nil { + return nil, nil + } + if depth > maxConversionDepth { + return nil, fmt.Errorf("output schema conversion exceeds maximum depth %d", maxConversionDepth) + } + c.nodes++ + if c.nodes > maxConversionNodes { + return nil, fmt.Errorf("output schema conversion exceeds maximum node count %d", maxConversionNodes) + } + if _, recursive := stack[schema]; recursive { + return nil, fmt.Errorf("recursive output schema is not supported") + } + next := make(map[*jsonschema.Schema]struct{}, len(stack)+1) + for item := range stack { + next[item] = struct{}{} + } + next[schema] = struct{}{} + + if schema.Ref != "" { + const prefix = "#/$defs/" + name := strings.TrimPrefix(schema.Ref, prefix) + if name == schema.Ref || strings.Contains(name, "/") { + return nil, fmt.Errorf("unsupported output schema reference %q", schema.Ref) + } + name = strings.ReplaceAll(strings.ReplaceAll(name, "~1", "/"), "~0", "~") + target := c.root.Defs[name] + if target == nil { + return nil, fmt.Errorf("output schema reference %q not found", schema.Ref) + } + return c.convert(target, next, depth+1) + } + + converted := &genai.Schema{Title: schema.Title, Description: schema.Description} + switch schema.Type { + case "": + case "object": + converted.Type = genai.TypeObject + case "array": + converted.Type = genai.TypeArray + case "string": + converted.Type = genai.TypeString + case "number": + converted.Type = genai.TypeNumber + case "integer": + converted.Type = genai.TypeInteger + case "boolean": + converted.Type = genai.TypeBoolean + case "null": + converted.Type = genai.TypeNULL + default: + return nil, fmt.Errorf("unsupported output schema type %q", schema.Type) + } + converted.Required = append([]string(nil), schema.Required...) + if schema.Items != nil { + items, err := c.convert(schema.Items, next, depth+1) + if err != nil { + return nil, err + } + converted.Items = items + } + if len(schema.Properties) > 0 { + converted.Properties = make(map[string]*genai.Schema, len(schema.Properties)) + for name, property := range schema.Properties { + value, err := c.convert(property, next, depth+1) + if err != nil { + return nil, fmt.Errorf("convert output property %q: %w", name, err) + } + converted.Properties[name] = value + } + } + for _, candidate := range schema.AnyOf { + value, err := c.convert(candidate, next, depth+1) + if err != nil { + return nil, fmt.Errorf("convert output anyOf: %w", err) + } + converted.AnyOf = append(converted.AnyOf, value) + } + convertedEnum := make([]string, 0, len(schema.Enum)) + for _, value := range schema.Enum { + text, ok := value.(string) + if !ok { + // genai.Schema can only express string enums. Omitting a mixed or + // non-string enum keeps the provider schema broader than the canonical + // contract; final response validation still enforces the full enum. + convertedEnum = nil + break + } + convertedEnum = append(convertedEnum, text) + } + converted.Enum = convertedEnum + if schema.Const != nil { + if value, ok := (*schema.Const).(string); ok { + converted.Enum = []string{value} + } + } + return converted, nil +} diff --git a/go/adk/pkg/outputschema/convert_test.go b/go/adk/pkg/outputschema/convert_test.go new file mode 100644 index 0000000000..34221c46ed --- /dev/null +++ b/go/adk/pkg/outputschema/convert_test.go @@ -0,0 +1,71 @@ +package outputschema + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/require" + "google.golang.org/genai" +) + +func TestToGenAISchema(t *testing.T) { + schema, err := ToGenAISchema(json.RawMessage(`{ + "type":"object", + "properties":{ + "status":{"$ref":"#/$defs/status"}, + "payload":{"type":"array","items":{"type":"integer"}} + }, + "required":["status"], + "$defs":{"status":{"type":"string","const":"success"}} + }`)) + require.NoError(t, err) + require.Equal(t, genai.TypeObject, schema.Type) + require.Equal(t, []string{"status"}, schema.Required) + require.Equal(t, []string{"success"}, schema.Properties["status"].Enum) + require.Equal(t, genai.TypeInteger, schema.Properties["payload"].Items.Type) +} + +func TestToGenAISchemaRejectsRecursiveReference(t *testing.T) { + _, err := ToGenAISchema(json.RawMessage(`{ + "type":"object", + "$defs":{"node":{"type":"object","properties":{"next":{"$ref":"#/$defs/node"}}}}, + "properties":{"node":{"$ref":"#/$defs/node"}} + }`)) + require.ErrorContains(t, err, "recursive output schema") +} + +func TestToGenAISchemaDoesNotNarrowUnsupportedConstraints(t *testing.T) { + schema, err := ToGenAISchema(json.RawMessage(`{ + "type":"object", + "properties":{ + "mixed":{"enum":["a",1]}, + "numeric":{"type":"integer","const":1} + }, + "additionalProperties":false + }`)) + require.NoError(t, err) + require.Empty(t, schema.Properties["mixed"].Enum) + require.Empty(t, schema.Properties["numeric"].Enum) +} + +func TestToGenAISchemaBoundsReferenceExpansion(t *testing.T) { + leafProperties := make(map[string]any, 30) + for i := range 30 { + leafProperties[string(rune('a'+i))] = map[string]any{"type": "string"} + } + rootProperties := make(map[string]any, 40) + for i := range 40 { + rootProperties[string(rune('A'+i))] = map[string]any{"$ref": "#/$defs/leaf"} + } + raw, err := json.Marshal(map[string]any{ + "type": "object", + "properties": rootProperties, + "$defs": map[string]any{ + "leaf": map[string]any{"type": "object", "properties": leafProperties}, + }, + }) + require.NoError(t, err) + + _, err = ToGenAISchema(raw) + require.ErrorContains(t, err, "conversion exceeds maximum node count") +} diff --git a/go/adk/pkg/tools/remote_a2a_tool.go b/go/adk/pkg/tools/remote_a2a_tool.go index 6ec6baf955..13228c9ef0 100644 --- a/go/adk/pkg/tools/remote_a2a_tool.go +++ b/go/adk/pkg/tools/remote_a2a_tool.go @@ -14,6 +14,7 @@ import ( "github.com/a2aproject/a2a-go/v2/a2asrv" "github.com/kagent-dev/kagent/go/adk/pkg/a2a" "github.com/kagent-dev/kagent/go/adk/pkg/constants" + kagenta2a "github.com/kagent-dev/kagent/go/api/a2a" "github.com/kagent-dev/kagent/go/pkg/logging" "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp" adkagent "google.golang.org/adk/v2/agent" @@ -407,8 +408,12 @@ func (s *remoteA2AState) handleResume(ctx adkagent.Context) (remoteA2AResponse, func (s *remoteA2AState) processResult(ctx adkagent.Context, contextID string, result a2atype.SendMessageResult) (remoteA2AResponse, error) { switch r := result.(type) { case *a2atype.Message: + text, err := extractTextFromMessage(r) + if err != nil { + return remoteA2AResponse{}, fmt.Errorf("extract remote agent message result: %w", err) + } return remoteA2AResponse{ - Result: extractTextFromMessage(r), + Result: text, SubagentSessionID: contextID, }, nil case *a2atype.Task: @@ -416,7 +421,10 @@ func (s *remoteA2AState) processResult(ctx adkagent.Context, contextID string, r case a2atype.TaskStateInputRequired: return s.handleInputRequired(ctx, r, contextID), nil case a2atype.TaskStateFailed: - text := extractTextFromTask(r) + text, err := extractTextFromTask(r) + if err != nil { + return remoteA2AResponse{}, fmt.Errorf("extract remote agent failure: %w", err) + } if text == "" { text = fmt.Sprintf("Remote agent '%s' failed.", s.name) } @@ -428,8 +436,12 @@ func (s *remoteA2AState) processResult(ctx adkagent.Context, contextID string, r // completed — include sub-agent's final LLM usage from task.metadata // so the parent can display it on the AgentCall card in the UI. // Mirrors Python's _extract_usage_from_task(task). + text, err := extractTextFromTask(r) + if err != nil { + return remoteA2AResponse{}, fmt.Errorf("extract remote agent task result: %w", err) + } ret := remoteA2AResponse{ - Result: extractTextFromTask(r), + Result: text, SubagentSessionID: contextID, } if usage := extractUsageFromTask(r); usage != nil { @@ -509,10 +521,10 @@ func extractUsageFromTask(task *a2atype.Task) map[string]any { return nil } -// extractTextFromTask extracts the text result from a completed Task. -func extractTextFromTask(task *a2atype.Task) string { +// extractTextFromTask extracts text or a structured terminal result from a completed Task. +func extractTextFromTask(task *a2atype.Task) (string, error) { if task == nil { - return "" + return "", nil } // Prefer artifacts (canonical result). if len(task.Artifacts) > 0 { @@ -524,24 +536,33 @@ func extractTextFromTask(task *a2atype.Task) string { } if text := part.Text(); text != "" { texts = append(texts, text) + continue + } + if !kagenta2a.IsStructuredOutputPart(part) { + continue } + text, err := kagenta2a.StructuredOutputJSON(part) + if err != nil { + return "", err + } + texts = append(texts, text) } } if len(texts) > 0 { - return strings.Join(texts, "\n") + return strings.Join(texts, "\n"), nil } } // Fall back to status message. if task.Status.Message != nil { return extractTextFromMessage(task.Status.Message) } - return "" + return "", nil } -// extractTextFromMessage extracts text from a direct A2A Message response. -func extractTextFromMessage(message *a2atype.Message) string { +// extractTextFromMessage extracts text or a structured result from a direct A2A Message response. +func extractTextFromMessage(message *a2atype.Message) (string, error) { if message == nil { - return "" + return "", nil } var texts []string for _, part := range message.Parts { @@ -550,7 +571,16 @@ func extractTextFromMessage(message *a2atype.Message) string { } if text := part.Text(); text != "" { texts = append(texts, text) + continue + } + if !kagenta2a.IsStructuredOutputPart(part) { + continue + } + text, err := kagenta2a.StructuredOutputJSON(part) + if err != nil { + return "", err } + texts = append(texts, text) } - return strings.Join(texts, "\n") + return strings.Join(texts, "\n"), nil } diff --git a/go/adk/pkg/tools/remote_a2a_tool_test.go b/go/adk/pkg/tools/remote_a2a_tool_test.go index 84b59c0e81..6443161a47 100644 --- a/go/adk/pkg/tools/remote_a2a_tool_test.go +++ b/go/adk/pkg/tools/remote_a2a_tool_test.go @@ -198,6 +198,21 @@ func TestProcessResult_SetsSubagentSessionIDOnEveryBranch(t *testing.T) { } }) + t.Run("completed Task structured result", func(t *testing.T) { + part := apia2a.NewStructuredOutputPart(map[string]any{"answer": float64(4)}, "digest") + task := &a2atype.Task{ + Status: a2atype.TaskStatus{State: a2atype.TaskStateCompleted}, + Artifacts: []*a2atype.Artifact{{Parts: a2atype.ContentParts{part}}}, + } + resp, err := s.processResult(nil, contextID, task) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if resp.Result != `{"answer":4}` { + t.Fatalf("Result = %q, want structured JSON", resp.Result) + } + }) + t.Run("unrecognised result type", func(t *testing.T) { resp, err := s.processResult(nil, contextID, nil) if err != nil { diff --git a/go/api/a2a/structured_output.go b/go/api/a2a/structured_output.go new file mode 100644 index 0000000000..7899f7c0bf --- /dev/null +++ b/go/api/a2a/structured_output.go @@ -0,0 +1,57 @@ +package a2a + +import ( + "encoding/json" + "fmt" + "strings" + + a2atype "github.com/a2aproject/a2a-go/v2/a2a" +) + +const outputSchemaSHA256MetadataKey = "kagent.dev/a2a/output-schema-sha256" + +// NewStructuredOutputPart constructs a structured terminal result with its +// complete wire signature: JSON data and the digest of the enforced schema. +func NewStructuredOutputPart(value any, schemaSHA256 string) *a2atype.Part { + part := a2atype.NewDataPart(value) + part.MediaType = "application/json" + part.SetMeta(outputSchemaSHA256MetadataKey, schemaSHA256) + return part +} + +// StructuredOutputSchemaSHA256 returns the digest carried by a structured +// terminal result. Ordinary data parts and incomplete signatures are rejected. +func StructuredOutputSchemaSHA256(part *a2atype.Part) (string, bool) { + if part == nil { + return "", false + } + if _, ok := part.Content.(a2atype.Data); !ok { + return "", false + } + mediaType, _, _ := strings.Cut(part.MediaType, ";") + if !strings.EqualFold(strings.TrimSpace(mediaType), "application/json") { + return "", false + } + digest, ok := part.Metadata[outputSchemaSHA256MetadataKey].(string) + return digest, ok +} + +// IsStructuredOutputPart reports whether a part carries kagent's structured +// terminal-result signature. +func IsStructuredOutputPart(part *a2atype.Part) bool { + _, ok := StructuredOutputSchemaSHA256(part) + return ok +} + +// StructuredOutputJSON serializes a structured terminal result for consumers +// whose output contract is text, such as the CLI and remote-agent tools. +func StructuredOutputJSON(part *a2atype.Part) (string, error) { + if !IsStructuredOutputPart(part) { + return "", fmt.Errorf("part is not structured output") + } + encoded, err := json.Marshal(part.Data()) + if err != nil { + return "", fmt.Errorf("marshal structured output: %w", err) + } + return string(encoded), nil +} diff --git a/go/api/a2a/structured_output_test.go b/go/api/a2a/structured_output_test.go new file mode 100644 index 0000000000..5756099a05 --- /dev/null +++ b/go/api/a2a/structured_output_test.go @@ -0,0 +1,31 @@ +package a2a + +import ( + "testing" + + a2atype "github.com/a2aproject/a2a-go/v2/a2a" + "github.com/stretchr/testify/require" +) + +func TestStructuredOutputJSON(t *testing.T) { + part := NewStructuredOutputPart(map[string]any{"answer": float64(4)}, "digest") + require.Equal(t, "application/json", part.MediaType) + digest, ok := StructuredOutputSchemaSHA256(part) + require.True(t, ok) + require.Equal(t, "digest", digest) + + part.MediaType = "application/json; charset=utf-8" + + text, err := StructuredOutputJSON(part) + require.NoError(t, err) + require.JSONEq(t, `{"answer":4}`, text) +} + +func TestStructuredOutputJSONRejectsOrdinaryDataParts(t *testing.T) { + part := a2atype.NewDataPart(map[string]any{"name": "tool", "response": "done"}) + part.MediaType = "application/json" + + text, err := StructuredOutputJSON(part) + require.EqualError(t, err, "part is not structured output") + require.Empty(t, text) +} diff --git a/go/api/adk/types.go b/go/api/adk/types.go index 7ed04e0296..5ab3786a4b 100644 --- a/go/api/adk/types.go +++ b/go/api/adk/types.go @@ -625,6 +625,14 @@ func (c *AgentCompressionConfig) UnmarshalJSON(data []byte) error { return nil } +// OutputConfig is the root agent's structured-output contract. JSONSchema is +// canonical JSON produced by the controller; SHA256 identifies that exact +// contract in public A2A results. +type OutputConfig struct { + JSONSchema json.RawMessage `json:"json_schema"` + SHA256 string `json:"sha256"` +} + // See `python/packages/kagent-adk/src/kagent/adk/types.py` for the python version of this type AgentConfig struct { Name string `json:"name,omitempty"` @@ -644,6 +652,7 @@ type AgentConfig struct { SessionDBURL string `json:"session_db_url,omitempty"` SkillsDirectory string `json:"skills_directory,omitempty"` SubAgents []*AgentConfig `json:"sub_agents,omitempty"` + Output *OutputConfig `json:"output,omitempty"` } // GetStream returns the stream value or default if not set @@ -673,6 +682,7 @@ func (a *AgentConfig) UnmarshalJSON(data []byte) error { SessionDBURL string `json:"session_db_url,omitempty"` SkillsDirectory string `json:"skills_directory,omitempty"` SubAgents []*AgentConfig `json:"sub_agents,omitempty"` + Output *OutputConfig `json:"output,omitempty"` } if err := json.Unmarshal(data, &tmp); err != nil { return err @@ -714,6 +724,7 @@ func (a *AgentConfig) UnmarshalJSON(data []byte) error { a.SessionDBURL = tmp.SessionDBURL a.SkillsDirectory = tmp.SkillsDirectory a.SubAgents = tmp.SubAgents + a.Output = tmp.Output return nil } diff --git a/go/api/config/crd/bases/kagent.dev_agenttemplates.yaml b/go/api/config/crd/bases/kagent.dev_agenttemplates.yaml index 8738a15588..eb52e4a103 100644 --- a/go/api/config/crd/bases/kagent.dev_agenttemplates.yaml +++ b/go/api/config/crd/bases/kagent.dev_agenttemplates.yaml @@ -65,6 +65,27 @@ spec: x-kubernetes-validations: - message: name must not be empty rule: has(self.name) && self.name != '' + outputSchema: + description: |- + OutputSchema constrains successful terminal output when this template is + compiled as the root agent. + type: object + x-kubernetes-preserve-unknown-fields: true + outputSchemaFrom: + description: |- + OutputSchemaFrom references a JSON Schema stored as JSON in a + same-namespace ConfigMap key. + properties: + key: + minLength: 1 + type: string + name: + minLength: 1 + type: string + required: + - key + - name + type: object plugins: items: description: PluginBundle selects Agent Skills from one immutable @@ -394,6 +415,8 @@ spec: x-kubernetes-validations: - message: systemPrompt and systemPromptFrom are mutually exclusive rule: '!(has(self.systemPrompt) && has(self.systemPromptFrom))' + - message: outputSchema and outputSchemaFrom are mutually exclusive + rule: '!(has(self.outputSchema) && has(self.outputSchemaFrom))' status: description: AgentTemplateStatus is the controller-observed state for each admitting Harness. diff --git a/go/api/config/crd/bases/kagent.dev_harnesses.yaml b/go/api/config/crd/bases/kagent.dev_harnesses.yaml index dd1868e807..b186ddabda 100644 --- a/go/api/config/crd/bases/kagent.dev_harnesses.yaml +++ b/go/api/config/crd/bases/kagent.dev_harnesses.yaml @@ -385,6 +385,11 @@ spec: type: boolean streaming: type: boolean + structuredOutput: + description: |- + StructuredOutput reports whether the pinned adapter can enforce a root + JSON output contract and emit it as an A2A DataPart. + type: boolean version: description: Version identifies the controller capability catalog entry. @@ -401,6 +406,7 @@ spec: - nativeAgentTools - resume - streaming + - structuredOutput - version type: object conditions: diff --git a/go/api/v1alpha3/agenttemplate_types.go b/go/api/v1alpha3/agenttemplate_types.go index 2f29a911d2..38be377e59 100644 --- a/go/api/v1alpha3/agenttemplate_types.go +++ b/go/api/v1alpha3/agenttemplate_types.go @@ -18,20 +18,11 @@ package v1alpha3 import ( corev1 "k8s.io/api/core/v1" + apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" ) -// AgentTemplateConfigMapKeyReference identifies a key in a same-namespace ConfigMap. -type AgentTemplateConfigMapKeyReference struct { - // +kubebuilder:validation:MinLength=1 - // +required - Name string `json:"name"` - // +kubebuilder:validation:MinLength=1 - // +required - Key string `json:"key"` -} - // AgentTemplatePromptTemplateSpec enables Go template rendering and ConfigMap includes. type AgentTemplatePromptTemplateSpec struct { // DataSources are same-namespace ConfigMaps available to include("source/key"). @@ -187,6 +178,7 @@ type PluginBundle struct { // AgentTemplateSpec defines portable agent behavior. // +kubebuilder:validation:XValidation:rule="!(has(self.systemPrompt) && has(self.systemPromptFrom))",message="systemPrompt and systemPromptFrom are mutually exclusive" +// +kubebuilder:validation:XValidation:rule="!(has(self.outputSchema) && has(self.outputSchemaFrom))",message="outputSchema and outputSchemaFrom are mutually exclusive" type AgentTemplateSpec struct { // ModelConfig is required by managed harnesses and optional for BYO harnesses. // +kubebuilder:validation:XValidation:rule="has(self.name) && self.name != ''",message="name must not be empty" @@ -198,7 +190,17 @@ type AgentTemplateSpec struct { SystemPrompt string `json:"systemPrompt,omitempty"` // SystemPromptFrom references prompt text in a same-namespace ConfigMap. // +optional - SystemPromptFrom *AgentTemplateConfigMapKeyReference `json:"systemPromptFrom,omitempty"` + SystemPromptFrom *ConfigMapKeyReference `json:"systemPromptFrom,omitempty"` + // OutputSchema constrains successful terminal output when this template is + // compiled as the root agent. + // +optional + // +kubebuilder:validation:Type=object + // +kubebuilder:pruning:PreserveUnknownFields + OutputSchema *apiextensionsv1.JSON `json:"outputSchema,omitempty"` + // OutputSchemaFrom references a JSON Schema stored as JSON in a + // same-namespace ConfigMap key. + // +optional + OutputSchemaFrom *ConfigMapKeyReference `json:"outputSchemaFrom,omitempty"` // +optional PromptTemplate *AgentTemplatePromptTemplateSpec `json:"promptTemplate,omitempty"` // +kubebuilder:validation:MaxItems=50 diff --git a/go/api/v1alpha3/common_types.go b/go/api/v1alpha3/common_types.go index 4113badce8..aa2bb49a1b 100644 --- a/go/api/v1alpha3/common_types.go +++ b/go/api/v1alpha3/common_types.go @@ -29,6 +29,16 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client" ) +// ConfigMapKeyReference identifies a key in a same-namespace ConfigMap. +type ConfigMapKeyReference struct { + // +kubebuilder:validation:MinLength=1 + // +required + Name string `json:"name"` + // +kubebuilder:validation:MinLength=1 + // +required + Key string `json:"key"` +} + // FromNamespaces specifies namespace from which references to this resource are allowed. // This follows the same pattern as Gateway API's cross-namespace route attachment. // See: https://gateway-api.sigs.k8s.io/guides/multiple-ns/#cross-namespace-route-attachment diff --git a/go/api/v1alpha3/harness_types.go b/go/api/v1alpha3/harness_types.go index ea223d7cb7..c6e681d200 100644 --- a/go/api/v1alpha3/harness_types.go +++ b/go/api/v1alpha3/harness_types.go @@ -237,6 +237,10 @@ type HarnessCapabilities struct { InputRequired bool `json:"inputRequired"` // +required Approvals bool `json:"approvals"` + // StructuredOutput reports whether the pinned adapter can enforce a root + // JSON output contract and emit it as an A2A DataPart. + // +required + StructuredOutput bool `json:"structuredOutput"` // +kubebuilder:validation:MaxItems=16 // +listType=set diff --git a/go/api/v1alpha3/zz_generated.deepcopy.go b/go/api/v1alpha3/zz_generated.deepcopy.go index 43836be741..74c5433c7f 100644 --- a/go/api/v1alpha3/zz_generated.deepcopy.go +++ b/go/api/v1alpha3/zz_generated.deepcopy.go @@ -54,21 +54,6 @@ func (in *AgentTemplate) DeepCopyObject() runtime.Object { return nil } -// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. -func (in *AgentTemplateConfigMapKeyReference) DeepCopyInto(out *AgentTemplateConfigMapKeyReference) { - *out = *in -} - -// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new AgentTemplateConfigMapKeyReference. -func (in *AgentTemplateConfigMapKeyReference) DeepCopy() *AgentTemplateConfigMapKeyReference { - if in == nil { - return nil - } - out := new(AgentTemplateConfigMapKeyReference) - in.DeepCopyInto(out) - return out -} - // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *AgentTemplateHarnessStatus) DeepCopyInto(out *AgentTemplateHarnessStatus) { *out = *in @@ -189,7 +174,17 @@ func (in *AgentTemplateSpec) DeepCopyInto(out *AgentTemplateSpec) { } if in.SystemPromptFrom != nil { in, out := &in.SystemPromptFrom, &out.SystemPromptFrom - *out = new(AgentTemplateConfigMapKeyReference) + *out = new(ConfigMapKeyReference) + **out = **in + } + if in.OutputSchema != nil { + in, out := &in.OutputSchema, &out.OutputSchema + *out = new(apiextensionsv1.JSON) + (*in).DeepCopyInto(*out) + } + if in.OutputSchemaFrom != nil { + in, out := &in.OutputSchemaFrom, &out.OutputSchemaFrom + *out = new(ConfigMapKeyReference) **out = **in } if in.PromptTemplate != nil { @@ -495,6 +490,21 @@ func (in *CodexHarness) DeepCopy() *CodexHarness { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *ConfigMapKeyReference) DeepCopyInto(out *ConfigMapKeyReference) { + *out = *in +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ConfigMapKeyReference. +func (in *ConfigMapKeyReference) DeepCopy() *ConfigMapKeyReference { + if in == nil { + return nil + } + out := new(ConfigMapKeyReference) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *FoundryConfig) DeepCopyInto(out *FoundryConfig) { *out = *in diff --git a/go/core/cli/internal/a2a/assembler.go b/go/core/cli/internal/a2a/assembler.go index 6d114fc580..3bfcf1904c 100644 --- a/go/core/cli/internal/a2a/assembler.go +++ b/go/core/cli/internal/a2a/assembler.go @@ -7,6 +7,7 @@ import ( a2atype "github.com/a2aproject/a2a-go/v2/a2a" "github.com/a2aproject/a2a-go/v2/a2aevent" + kagenta2a "github.com/kagent-dev/kagent/go/api/a2a" ) // Assembler projects an A2A event stream into its current Message or Task. @@ -73,13 +74,26 @@ func (a *Assembler) Complete() bool { return state.Terminal() || state == a2atype.TaskStateInputRequired || state == a2atype.TaskStateAuthRequired } -// PartsText concatenates the text of every text part, ignoring data parts. -func PartsText(parts a2atype.ContentParts) string { +// PartsText concatenates text parts and serializes structured terminal results. +// Other data parts are runtime activity and remain excluded from answer text. +func PartsText(parts a2atype.ContentParts) (string, error) { var text strings.Builder for _, part := range parts { - if part != nil { - text.WriteString(part.Text()) + if part == nil { + continue } + if value := part.Text(); value != "" { + text.WriteString(value) + continue + } + if !kagenta2a.IsStructuredOutputPart(part) { + continue + } + value, err := kagenta2a.StructuredOutputJSON(part) + if err != nil { + return "", err + } + text.WriteString(value) } - return text.String() + return text.String(), nil } diff --git a/go/core/cli/internal/a2a/assembler_test.go b/go/core/cli/internal/a2a/assembler_test.go index a4bcdb3598..3baefbe639 100644 --- a/go/core/cli/internal/a2a/assembler_test.go +++ b/go/core/cli/internal/a2a/assembler_test.go @@ -4,6 +4,7 @@ import ( "testing" a2atype "github.com/a2aproject/a2a-go/v2/a2a" + kagenta2a "github.com/kagent-dev/kagent/go/api/a2a" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -56,3 +57,12 @@ func TestAssemblerAcceptsMessageResult(t *testing.T) { assert.Same(t, message, assembler.Result()) assert.True(t, assembler.Complete()) } + +func TestPartsTextSerializesStructuredOutput(t *testing.T) { + structured := kagenta2a.NewStructuredOutputPart(map[string]any{"answer": float64(4)}, "digest") + toolResult := a2atype.NewDataPart(map[string]any{"name": "tool", "response": "ignored"}) + + text, err := PartsText(a2atype.ContentParts{toolResult, structured}) + require.NoError(t, err) + require.JSONEq(t, `{"answer":4}`, text) +} diff --git a/go/core/cli/internal/commands/agentinstance/invoke.go b/go/core/cli/internal/commands/agentinstance/invoke.go index 168b054ab7..4e7b49ea4b 100644 --- a/go/core/cli/internal/commands/agentinstance/invoke.go +++ b/go/core/cli/internal/commands/agentinstance/invoke.go @@ -205,7 +205,10 @@ type tableStreamWriter struct { } func (w *tableStreamWriter) Write(_ a2atype.Event, result a2atype.SendMessageResult) error { - text := sendResultText(result) + text, err := sendResultText(result) + if err != nil { + return err + } if text == w.text { return nil } @@ -245,7 +248,10 @@ func writeSendResult(w io.Writer, format clioutput.Format, result a2atype.SendMe } func writeTableResult(w io.Writer, result a2atype.SendMessageResult) error { - text := sendResultText(result) + text, err := sendResultText(result) + if err != nil { + return err + } if text != "" { if _, err := io.WriteString(w, text); err != nil { return fmt.Errorf("write invoke output: %w", err) @@ -280,32 +286,40 @@ func writeTableContinuation(w io.Writer, result a2atype.SendMessageResult) error return nil } -func sendResultText(result a2atype.SendMessageResult) string { +func sendResultText(result a2atype.SendMessageResult) (string, error) { switch result := result.(type) { case *a2atype.Message: return messageText(result) case *a2atype.Task: groups := make([]string, 0, len(result.Artifacts)+1) - if text := messageText(result.Status.Message); text != "" { - groups = append(groups, text) + statusText, err := messageText(result.Status.Message) + if err != nil { + return "", err + } + if statusText != "" { + groups = append(groups, statusText) } for _, artifact := range result.Artifacts { if artifact == nil { continue } - if text := clia2a.PartsText(artifact.Parts); text != "" { + text, err := clia2a.PartsText(artifact.Parts) + if err != nil { + return "", err + } + if text != "" { groups = append(groups, text) } } - return strings.Join(groups, "\n") + return strings.Join(groups, "\n"), nil default: - return "" + return "", nil } } -func messageText(message *a2atype.Message) string { +func messageText(message *a2atype.Message) (string, error) { if message == nil { - return "" + return "", nil } return clia2a.PartsText(message.Parts) } diff --git a/go/core/cli/internal/commands/agentinstance/invoke_test.go b/go/core/cli/internal/commands/agentinstance/invoke_test.go index 18e3b50380..64c97063b8 100644 --- a/go/core/cli/internal/commands/agentinstance/invoke_test.go +++ b/go/core/cli/internal/commands/agentinstance/invoke_test.go @@ -232,7 +232,9 @@ func TestConsumeA2AStreamReturnsPartialResultOnTruncation(t *testing.T) { ), func(a2atype.Event, a2atype.SendMessageResult) error { return nil }) require.ErrorIs(t, err, errTruncatedA2AStream) - assert.Equal(t, "partial", sendResultText(result)) + text, err := sendResultText(result) + require.NoError(t, err) + assert.Equal(t, "partial", text) } func TestConsumeA2AStreamWritesJSONL(t *testing.T) { @@ -254,7 +256,9 @@ func TestConsumeA2AStreamWritesJSONL(t *testing.T) { return writeStreamEvent(&output, event) }) require.NoError(t, err) - assert.Equal(t, "done", sendResultText(result)) + text, err := sendResultText(result) + require.NoError(t, err) + assert.Equal(t, "done", text) lines := strings.Split(strings.TrimSpace(output.String()), "\n") require.Len(t, lines, len(events)) for _, line := range lines { diff --git a/go/core/cli/internal/tui/chat.go b/go/core/cli/internal/tui/chat.go index f3f25f506c..3d8d98b928 100644 --- a/go/core/cli/internal/tui/chat.go +++ b/go/core/cli/internal/tui/chat.go @@ -306,7 +306,11 @@ func eventParts(ev a2atype.Event) a2atype.ContentParts { // renderAssembledText appends newly assembled text; the cumulative projection grows the block in place. func (m *chatModel) renderAssembledText() { - text := assembledText(m.assembler.Result()) + text, err := assembledText(m.assembler.Result()) + if err != nil { + m.appendTransportError(err) + return + } if text == m.projected { return } @@ -322,7 +326,7 @@ func (m *chatModel) renderAssembledText() { } // assembledText projects agent output; only artifacts carry it, status messages are control-plane. -func assembledText(result a2atype.SendMessageResult) string { +func assembledText(result a2atype.SendMessageResult) (string, error) { switch result := result.(type) { case *a2atype.Message: return clia2a.PartsText(result.Parts) @@ -332,13 +336,17 @@ func assembledText(result a2atype.SendMessageResult) string { if artifact == nil { continue } - if text := clia2a.PartsText(artifact.Parts); text != "" { + text, err := clia2a.PartsText(artifact.Parts) + if err != nil { + return "", err + } + if text != "" { groups = append(groups, text) } } - return strings.Join(groups, "\n") + return strings.Join(groups, "\n"), nil default: - return "" + return "", nil } } @@ -363,7 +371,10 @@ func (m *chatModel) renderState() { case a2atype.TaskStateFailed, a2atype.TaskStateRejected, a2atype.TaskStateCanceled: banner := fmt.Sprintf("✗ Task %s.", state) if task.Status.Message != nil { - if detail := clia2a.PartsText(task.Status.Message.Parts); strings.TrimSpace(detail) != "" { + detail, err := clia2a.PartsText(task.Status.Message.Parts) + if err != nil { + m.appendTransportError(err) + } else if strings.TrimSpace(detail) != "" { banner += " " + detail } } @@ -386,11 +397,21 @@ func (m *chatModel) AppendHistoryTask(task *a2atype.Task) { if msg == nil || msg.Role != a2atype.MessageRoleUser { continue } - if text := clia2a.PartsText(msg.Parts); strings.TrimSpace(text) != "" { + text, err := clia2a.PartsText(msg.Parts) + if err != nil { + m.appendTransportError(err) + continue + } + if strings.TrimSpace(text) != "" { m.appendUser(text) } } - if text := assembledText(task); strings.TrimSpace(text) != "" { + text, err := assembledText(task) + if err != nil { + m.appendTransportError(err) + return + } + if strings.TrimSpace(text) != "" { m.appendLine(theme.AgentStyle().Render("Agent:") + "\n" + text) } } diff --git a/go/core/internal/translator/adkconfig/builder.go b/go/core/internal/translator/adkconfig/builder.go index 187de8341f..8e799cbee0 100644 --- a/go/core/internal/translator/adkconfig/builder.go +++ b/go/core/internal/translator/adkconfig/builder.go @@ -212,6 +212,9 @@ func (c *Builder) BuildProvenance(ctx context.Context, harness *v1alpha3.Harness if template.Spec.SystemPromptFrom != nil { configMaps[template.Spec.SystemPromptFrom.Name] = struct{}{} } + if template.Spec.OutputSchemaFrom != nil { + configMaps[template.Spec.OutputSchemaFrom.Name] = struct{}{} + } if template.Spec.PromptTemplate != nil { for _, source := range template.Spec.PromptTemplate.DataSources { configMaps[source.Name] = struct{}{} diff --git a/go/core/internal/translator/agentcard.go b/go/core/internal/translator/agentcard.go index 920767da09..dbda964a4b 100644 --- a/go/core/internal/translator/agentcard.go +++ b/go/core/internal/translator/agentcard.go @@ -12,10 +12,14 @@ import ( // the kagent, Codex, and Claude harnesses. The gateway replaces the private // interface while preserving these runtime capabilities. func ManagedAgentCard(template *v1alpha3.AgentTemplate) *a2atype.AgentCard { + outputModes := []string{"text"} + if template.Spec.OutputSchema != nil || template.Spec.OutputSchemaFrom != nil { + outputModes = []string{"application/json"} + } return &a2atype.AgentCard{ Name: strings.ReplaceAll(template.Name, "-", "_"), Description: template.Spec.Description, Version: "v1", SupportedInterfaces: []*a2atype.AgentInterface{{URL: "http://127.0.0.1:80", ProtocolBinding: a2atype.TransportProtocolGRPC, ProtocolVersion: a2atype.Version}}, Capabilities: a2atype.AgentCapabilities{Streaming: true, Extensions: []a2atype.AgentExtension{apia2a.HITLExtension()}}, - Skills: []a2atype.AgentSkill{}, DefaultInputModes: []string{"text"}, DefaultOutputModes: []string{"text"}, + Skills: []a2atype.AgentSkill{}, DefaultInputModes: []string{"text"}, DefaultOutputModes: outputModes, } } diff --git a/go/core/internal/translator/compiler.go b/go/core/internal/translator/compiler.go index a35439a8e8..2852702fa6 100644 --- a/go/core/internal/translator/compiler.go +++ b/go/core/internal/translator/compiler.go @@ -59,8 +59,9 @@ type ResolvedAgentBinding struct { // HarnessInput contains the Kubernetes inputs needed by a harness compiler. type HarnessInput struct { - Harness *v1alpha3.Harness - Root *AgentInput + Harness *v1alpha3.Harness + Root *AgentInput + OutputSchema *ResolvedOutputSchema } // AgentInput contains resolved Kubernetes inputs for one agent. @@ -196,6 +197,13 @@ func harnessSelector(harness *v1alpha3.Harness) (labels.Selector, error) { } func (c *Compiler) buildInputs(ctx context.Context, tree *ResolvedTree) (*HarnessInput, error) { + outputSchema, err := c.resolveOutputSchema(ctx, tree.Root.Template) + if err != nil { + return nil, err + } + if outputSchema != nil && harnessType(tree.Harness) != HarnessTypeKagent { + return nil, NewValidationError("Harness %q does not support structured output", tree.Harness.Name) + } var build func(*ResolvedAgent) (*AgentInput, error) build = func(agent *ResolvedAgent) (*AgentInput, error) { template := agent.Template @@ -266,5 +274,5 @@ func (c *Compiler) buildInputs(ctx context.Context, tree *ResolvedTree) (*Harnes if err != nil { return nil, err } - return &HarnessInput{Harness: tree.Harness, Root: root}, nil + return &HarnessInput{Harness: tree.Harness, Root: root, OutputSchema: outputSchema}, nil } diff --git a/go/core/internal/translator/compiler_test.go b/go/core/internal/translator/compiler_test.go index 8baa1ec982..d2684527f1 100644 --- a/go/core/internal/translator/compiler_test.go +++ b/go/core/internal/translator/compiler_test.go @@ -16,6 +16,7 @@ import ( "istio.io/istio/pkg/kube/krt" "istio.io/istio/pkg/kube/krt/krttest" corev1 "k8s.io/api/core/v1" + apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" ) @@ -136,6 +137,52 @@ func mockCollections(t *testing.T, objects ...any) v2translator.Collections { return collections } +func TestCompileAgentTemplateStructuredOutput(t *testing.T) { + harness := &v1alpha3.Harness{ + ObjectMeta: metav1.ObjectMeta{Name: "kagent", Namespace: "test"}, + Spec: v1alpha3.HarnessSpec{ + Kagent: &v1alpha3.KagentHarness{}, + AllowedAgentTemplates: &v1alpha3.HarnessAgentTemplateAdmission{Selector: metav1.LabelSelector{ + MatchLabels: map[string]string{"runtime": "kagent"}, + }}, + }, + } + schema := `{"type":"object","properties":{"answer":{"type":"integer"}},"required":["answer"],"additionalProperties":false}` + configMap := &corev1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{Name: "schemas", Namespace: "test", UID: "schemas-uid"}, + Data: map[string]string{"answer.json": schema}, + } + child := &v1alpha3.AgentTemplate{ + ObjectMeta: metav1.ObjectMeta{Name: "child", Namespace: "test", Labels: map[string]string{"runtime": "kagent"}}, + Spec: v1alpha3.AgentTemplateSpec{ + ModelConfig: &corev1.LocalObjectReference{Name: "default-model"}, + // A child contract is intentionally not resolved while this template is nested. + OutputSchema: &apiextensionsv1.JSON{Raw: []byte(`{"type":"object","oneOf":[]}`)}, + }, + } + root := &v1alpha3.AgentTemplate{ + ObjectMeta: metav1.ObjectMeta{Name: "root", Namespace: "test", Labels: map[string]string{"runtime": "kagent"}}, + Spec: v1alpha3.AgentTemplateSpec{ + ModelConfig: &corev1.LocalObjectReference{Name: "default-model"}, + OutputSchemaFrom: &v1alpha3.ConfigMapKeyReference{Name: configMap.Name, Key: "answer.json"}, + Tools: []v1alpha3.ToolBinding{{Agent: &v1alpha3.AgentToolBinding{ + Name: "child", Description: "delegate", TemplateRef: corev1.LocalObjectReference{Name: child.Name}, + }}}, + }, + } + + revision, err := compiler(t, modelConfig(), configMap, child).CompileAgentTemplate(t.Context(), harness, root) + require.NoError(t, err) + var config adk.AgentConfig + require.NoError(t, json.Unmarshal(revision.ConfigJSON, &config)) + require.JSONEq(t, schema, string(config.Output.JSONSchema)) + require.Len(t, config.Output.SHA256, 64) + require.Len(t, config.SubAgents, 1) + require.Nil(t, config.SubAgents[0].Output) + require.Contains(t, string(revision.Provenance), `"kind":"ConfigMap"`) + require.Equal(t, []string{"application/json"}, revision.AgentCard.DefaultOutputModes) +} + func TestResolveModelConfigFoundryEndpoint(t *testing.T) { ref := &corev1.ConfigMapKeySelector{LocalObjectReference: corev1.LocalObjectReference{Name: "account"}, Key: "endpoint"} const endpoint = "https://example.services.ai.azure.com" @@ -212,6 +259,28 @@ func TestCompilerAcceptsExternalHarnessCompiler(t *testing.T) { require.Equal(t, modelConfig().Spec, adapter.input.Root.ResolvedModelConfig.Config.Spec) } +func TestCompilerRejectsStructuredOutputForUnsupportedHarness(t *testing.T) { + collections := mockCollections(t, modelConfig()) + adapter := &testHarnessCompiler{} + harness := &v1alpha3.Harness{ + ObjectMeta: metav1.ObjectMeta{Name: "codex", Namespace: "test"}, + Spec: v1alpha3.HarnessSpec{Codex: &v1alpha3.CodexHarness{}, AllowedAgentTemplates: &v1alpha3.HarnessAgentTemplateAdmission{Selector: metav1.LabelSelector{}}}, + } + template := &v1alpha3.AgentTemplate{ + ObjectMeta: metav1.ObjectMeta{Name: "assistant", Namespace: "test"}, + Spec: v1alpha3.AgentTemplateSpec{ + ModelConfig: &corev1.LocalObjectReference{Name: "default-model"}, + OutputSchema: &apiextensionsv1.JSON{Raw: []byte(`{"type":"object"}`)}, + }, + } + + _, err := v2translator.NewCompiler(krt.TestingDummyContext{}, collections, map[v2translator.HarnessType]v2translator.HarnessCompiler{ + v2translator.HarnessTypeCodex: adapter, + }).CompileAgentTemplate(context.Background(), harness, template) + require.ErrorContains(t, err, `Harness "codex" does not support structured output`) + require.Nil(t, adapter.input) +} + func TestCompilerRejectsUnusableModelConfigBeforeHarnessCompiler(t *testing.T) { model := modelConfig() model.Spec.APIKeySecret = "missing" diff --git a/go/core/internal/translator/kagent/compiler.go b/go/core/internal/translator/kagent/compiler.go index 62ba8c102c..066cd5f3ea 100644 --- a/go/core/internal/translator/kagent/compiler.go +++ b/go/core/internal/translator/kagent/compiler.go @@ -7,6 +7,7 @@ import ( "slices" "github.com/a2aproject/a2a-go/v2/a2apb/v1/pbconv" + adkoutputschema "github.com/kagent-dev/kagent/go/adk/pkg/outputschema" "github.com/kagent-dev/kagent/go/api/adk" v2translator "github.com/kagent-dev/kagent/go/core/internal/translator" "github.com/kagent-dev/kagent/go/core/internal/translator/adkconfig" @@ -37,6 +38,9 @@ func (c *Compiler) Compile(ctx context.Context, input *v2translator.HarnessInput if err != nil { return nil, err } + if err := applyOutputSchema(compiled.Config, input.OutputSchema); err != nil { + return nil, err + } template, harness := input.Root.Template, input.Harness if err := c.config.ApplyCompaction(compiled, harness, template); err != nil { return nil, err @@ -102,6 +106,24 @@ func (c *Compiler) Compile(ctx context.Context, input *v2translator.HarnessInput }}, nil } +// applyOutputSchema verifies the portable schema by performing the same +// genai.Schema conversion used by the Go runtime, then records the canonical +// schema for both Go and Python ADK runtimes. This keeps compatibility +// failures at Harness compilation instead of actor startup. +func applyOutputSchema(config *adk.AgentConfig, output *v2translator.ResolvedOutputSchema) error { + if output == nil { + return nil + } + if _, err := adkoutputschema.ToGenAISchema(output.Schema); err != nil { + return v2translator.NewValidationError("output schema is incompatible with Go ADK: %v", err) + } + config.Output = &adk.OutputConfig{ + JSONSchema: append(json.RawMessage(nil), output.Schema...), + SHA256: output.SHA256, + } + return nil +} + func requireModels(input *v2translator.AgentInput) error { if input.ResolvedModelConfig == nil || input.ResolvedModelConfig.Config == nil { return v2translator.NewValidationError("kagent ModelConfig is required") diff --git a/go/core/internal/translator/kagent/outputschema_test.go b/go/core/internal/translator/kagent/outputschema_test.go new file mode 100644 index 0000000000..435c162ff7 --- /dev/null +++ b/go/core/internal/translator/kagent/outputschema_test.go @@ -0,0 +1,34 @@ +package kagent + +import ( + "testing" + + "github.com/kagent-dev/kagent/go/api/adk" + v2translator "github.com/kagent-dev/kagent/go/core/internal/translator" + "github.com/stretchr/testify/require" +) + +func TestApplyOutputSchema(t *testing.T) { + config := &adk.AgentConfig{} + output := &v2translator.ResolvedOutputSchema{ + Schema: []byte(`{"type":"object","$defs":{"status":{"type":"string"}},"properties":{"status":{"$ref":"#/$defs/status"}}}`), + SHA256: "digest", + } + + require.NoError(t, applyOutputSchema(config, output)) + require.JSONEq(t, string(output.Schema), string(config.Output.JSONSchema)) + require.Equal(t, "digest", config.Output.SHA256) +} + +func TestApplyOutputSchemaRejectsRecursiveADKConversion(t *testing.T) { + config := &adk.AgentConfig{} + output := &v2translator.ResolvedOutputSchema{Schema: []byte(`{ + "type":"object", + "$defs":{"node":{"type":"object","properties":{"next":{"$ref":"#/$defs/node"}}}}, + "properties":{"node":{"$ref":"#/$defs/node"}} + }`)} + + err := applyOutputSchema(config, output) + require.ErrorContains(t, err, "recursive output schema") + require.Nil(t, config.Output) +} diff --git a/go/core/internal/translator/outputschema.go b/go/core/internal/translator/outputschema.go new file mode 100644 index 0000000000..9e97395f6d --- /dev/null +++ b/go/core/internal/translator/outputschema.go @@ -0,0 +1,157 @@ +package translator + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "sync" + + "github.com/google/jsonschema-go/jsonschema" + "github.com/kagent-dev/kagent/go/api/v1alpha3" + "istio.io/istio/pkg/kube/krt" + "k8s.io/apimachinery/pkg/types" +) + +const maxOutputSchemaBytes = 64 << 10 + +// portableOutputSchemaMetaSchema describes the JSON Schema subset that every +// supported kagent ADK integration can consume. It is deliberately itself a +// JSON Schema so keyword forms and recursive schema locations are validated by +// jsonschema-go rather than by a parallel handwritten schema implementation. +const portableOutputSchemaMetaSchema = `{ + "$schema":"https://json-schema.org/draft/2020-12/schema", + "$defs":{ + "schema":{ + "type":"object", + "properties":{ + "$id":{"type":"string"}, + "$schema":{"type":"string"}, + "$defs":{"type":"object","additionalProperties":{"$ref":"#/$defs/schema"}}, + "$ref":{"type":"string","pattern":"^#/\\$defs/[^/]*$"}, + "additionalProperties":{"anyOf":[{"type":"boolean"},{"$ref":"#/$defs/schema"}]}, + "anyOf":{"type":"array","minItems":1,"items":{"$ref":"#/$defs/schema"}}, + "const":{}, + "description":{"type":"string"}, + "enum":{"type":"array","minItems":1}, + "items":{"$ref":"#/$defs/schema"}, + "properties":{"type":"object","additionalProperties":{"$ref":"#/$defs/schema"}}, + "required":{"type":"array","items":{"type":"string","minLength":1},"uniqueItems":true}, + "title":{"type":"string"}, + "type":{"enum":["object","array","string","number","integer","boolean","null"]} + }, + "additionalProperties":false + } + }, + "allOf":[{"$ref":"#/$defs/schema"}], + "properties":{"type":{"const":"object"}}, + "required":["type"] +}` + +var resolvePortableOutputSchemaMetaSchema = sync.OnceValues(func() (*jsonschema.Resolved, error) { + var schema jsonschema.Schema + if err := json.Unmarshal([]byte(portableOutputSchemaMetaSchema), &schema); err != nil { + return nil, fmt.Errorf("decode portable output meta-schema: %w", err) + } + resolved, err := schema.Resolve(nil) + if err != nil { + return nil, fmt.Errorf("resolve portable output meta-schema: %w", err) + } + return resolved, nil +}) + +// ResolvedOutputSchema is the canonical root output contract passed to a +// Harness compiler. Source is nil for an inline schema. +type ResolvedOutputSchema struct { + Schema []byte + SHA256 string + Source *ResolvedConfigMapKeyReference +} + +// ResolvedConfigMapKeyReference identifies the exact ConfigMap input used by a +// compiled revision. +type ResolvedConfigMapKeyReference struct { + NamespacedName types.NamespacedName + Key string +} + +func (c *Compiler) resolveOutputSchema(_ context.Context, template *v1alpha3.AgentTemplate) (*ResolvedOutputSchema, error) { + var raw []byte + var source *ResolvedConfigMapKeyReference + switch { + case template.Spec.OutputSchema != nil: + raw = template.Spec.OutputSchema.Raw + case template.Spec.OutputSchemaFrom != nil: + ref := template.Spec.OutputSchemaFrom + key := types.NamespacedName{Namespace: template.Namespace, Name: ref.Name} + configMap := krt.FetchOne(c.ctx, c.collections.ConfigMaps, krt.FilterObjectName(key)) + if configMap == nil { + return nil, fmt.Errorf("resolve output schema ConfigMap %q: not found", ref.Name) + } + value, ok := (*configMap).Data[ref.Key] + if !ok { + return nil, fmt.Errorf("resolve output schema ConfigMap %q: key %q not found", ref.Name, ref.Key) + } + raw = []byte(value) + source = &ResolvedConfigMapKeyReference{NamespacedName: key, Key: ref.Key} + default: + return nil, nil + } + + canonical, err := validateOutputSchema(raw) + if err != nil { + return nil, NewValidationError("invalid output schema: %v", err) + } + digest := sha256.Sum256(canonical) + return &ResolvedOutputSchema{Schema: canonical, SHA256: hex.EncodeToString(digest[:]), Source: source}, nil +} + +// validateOutputSchema validates a document against kagent's portable JSON +// Schema profile, then asks jsonschema-go to parse and resolve the schema +// itself. Go ADK compatibility checks that depend on genai.Schema projection, +// such as reference expansion and cycle detection, run in the kagent Harness +// compiler. +func validateOutputSchema(raw []byte) ([]byte, error) { + if len(raw) == 0 { + return nil, fmt.Errorf("schema is empty") + } + if len(raw) > maxOutputSchemaBytes { + return nil, fmt.Errorf("schema exceeds %d bytes", maxOutputSchemaBytes) + } + decoder := json.NewDecoder(bytes.NewReader(raw)) + var document any + if err := decoder.Decode(&document); err != nil { + return nil, fmt.Errorf("decode JSON: %w", err) + } + var trailing any + if err := decoder.Decode(&trailing); err != io.EOF { + if err == nil { + return nil, fmt.Errorf("schema must contain one JSON value") + } + return nil, fmt.Errorf("decode trailing JSON: %w", err) + } + + profile, err := resolvePortableOutputSchemaMetaSchema() + if err != nil { + return nil, err + } + if err := profile.Validate(document); err != nil { + return nil, fmt.Errorf("schema does not match the portable output profile: %w", err) + } + + canonical, err := json.Marshal(document) + if err != nil { + return nil, fmt.Errorf("canonicalize schema: %w", err) + } + var schema jsonschema.Schema + if err := json.Unmarshal(canonical, &schema); err != nil { + return nil, fmt.Errorf("parse schema: %w", err) + } + if _, err := schema.Resolve(nil); err != nil { + return nil, fmt.Errorf("resolve schema: %w", err) + } + return canonical, nil +} diff --git a/go/core/internal/translator/outputschema_test.go b/go/core/internal/translator/outputschema_test.go new file mode 100644 index 0000000000..d6a502d7de --- /dev/null +++ b/go/core/internal/translator/outputschema_test.go @@ -0,0 +1,75 @@ +package translator + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestValidateOutputSchema(t *testing.T) { + tests := []struct { + name string + schema string + wantErr string + }{ + { + name: "portable object schema", + schema: `{ + "type":"object", + "properties":{"status":{"$ref":"#/$defs/status"},"payload":{"type":"object","additionalProperties":false}}, + "required":["status","payload"], + "additionalProperties":false, + "$defs":{"status":{"type":"string","enum":["success","error"]}} + }`, + }, + { + name: "standard schema identity keywords", + schema: `{"$schema":"https://json-schema.org/draft/2020-12/schema","$id":"https://example.com/result.schema.json","type":"object"}`, + }, + { + name: "all portable schema forms", + schema: `{ + "type":"object", + "title":"Result", + "description":"A structured result", + "properties":{ + "value":{"anyOf":[{"$ref":"#/$defs/value"},{"type":"null","const":null}]}, + "labels":{"type":"array","items":{"type":"string","enum":["one","two"]}}, + "metadata":{"type":"object","additionalProperties":{"type":"string"}} + }, + "required":["value","labels"], + "additionalProperties":false, + "$defs":{"value":{"type":"integer","enum":[1,2]}} + }`, + }, + {name: "root is not object", schema: `{"type":"array","items":{"type":"string"}}`, wantErr: "does not equal object"}, + {name: "unknown keyword", schema: `{"type":"object","oneOf":[]}`, wantErr: "unexpected additional properties"}, + {name: "external ref", schema: `{"type":"object","properties":{"x":{"$ref":"https://example.com/schema"}}}`, wantErr: "does not match regular expression"}, + {name: "missing ref", schema: `{"type":"object","properties":{"x":{"$ref":"#/$defs/missing"}}}`, wantErr: `no key "missing"`}, + {name: "duplicate required", schema: `{"type":"object","required":["x","x"]}`, wantErr: "uniqueItems"}, + {name: "recursive ref is valid before ADK projection", schema: `{"type":"object","$defs":{"node":{"type":"object","properties":{"next":{"$ref":"#/$defs/node"}}}},"properties":{"node":{"$ref":"#/$defs/node"}}}`}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + canonical, err := validateOutputSchema([]byte(test.schema)) + if test.wantErr != "" { + require.ErrorContains(t, err, test.wantErr) + return + } + require.NoError(t, err) + require.JSONEq(t, test.schema, string(canonical)) + }) + } +} + +func TestValidateOutputSchemaBoundsInputSize(t *testing.T) { + raw, err := json.Marshal(map[string]any{ + "type": "object", + "description": string(make([]byte, maxOutputSchemaBytes)), + }) + require.NoError(t, err) + + _, err = validateOutputSchema(raw) + require.ErrorContains(t, err, "schema exceeds") +} diff --git a/go/core/test/e2e/interaction_test.go b/go/core/test/e2e/interaction_test.go index 3f426c17b1..4d0233c57e 100644 --- a/go/core/test/e2e/interaction_test.go +++ b/go/core/test/e2e/interaction_test.go @@ -40,6 +40,7 @@ import ( "google.golang.org/grpc/status" corev1 "k8s.io/api/core/v1" discoveryv1 "k8s.io/api/discovery/v1" + apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" k8sruntime "k8s.io/apimachinery/pkg/runtime" @@ -48,9 +49,11 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client/config" ) -//go:embed mocks/invoke_golang_adk_agent.json mocks/invoke_golang_hitl_ask_user.json mocks/invoke_mcp_agent.json mocks/invoke_shared_agent.json +//go:embed mocks/invoke_golang_adk_agent.json mocks/invoke_golang_hitl_ask_user.json mocks/invoke_mcp_agent.json mocks/invoke_shared_agent.json mocks/invoke_structured_output.json var interactionMocks embed.FS +const structuredOutputSchema = `{"type":"object","properties":{"answer":{"type":"integer"},"explanation":{"type":"string"}},"required":["answer","explanation"],"additionalProperties":false}` + // TestAgentInstanceInteraction verifies the complete public interaction path: // gateway routing, Substrate Actor transport, Go ADK execution, and the model call. func TestAgentInstanceInteraction(t *testing.T) { @@ -71,6 +74,56 @@ func TestAgentInstanceInteraction(t *testing.T) { } } +func TestAgentInstanceStructuredOutput(t *testing.T) { + t.Parallel() + target := interactionTarget(t) + kube := interactionKubeClient(t) + mcpURL, mcpServer := startMCPMock(t) + template := createStructuredOutputInteractionTemplate(t, kube, startMockLLM(t, "mocks/invoke_structured_output.json"), mcpURL) + fixture := newInteractionFixtureForHarnessTemplate(t, target, "kagent", template.Name) + _, _, task := fixture.send(t, "Add 3 and 5 and return the structured result.") + assertStructuredOutputTask(t, task, 8, "three plus five equals eight") + + for _, request := range mcpServer.Requests() { + if bytes.Contains(request.Body, []byte(`"method":"tools/call"`)) && bytes.Contains(request.Body, []byte(`"name":"add_numbers"`)) { + return + } + } + t.Fatal("mock MCP server did not receive the add_numbers call used by the structured response") +} + +func assertStructuredOutputTask(t *testing.T, task *a2atype.Task, answer float64, explanation string) { + t.Helper() + if task.Status.State != a2atype.TaskStateCompleted { + t.Fatalf("A2A task state = %s, want COMPLETED", task.Status.State) + } + if len(task.Artifacts) == 0 { + t.Fatal("structured task has no result artifact") + } + assertStructuredOutputArtifact(t, task.Artifacts[len(task.Artifacts)-1], answer, explanation) +} + +func assertStructuredOutputArtifact(t *testing.T, artifact *a2atype.Artifact, answer float64, explanation string) { + t.Helper() + if artifact == nil { + t.Fatal("structured result artifact is nil") + } + if len(artifact.Parts) != 1 { + t.Fatalf("structured result has %d parts, want 1", len(artifact.Parts)) + } + part := artifact.Parts[0] + data, ok := part.Data().(map[string]any) + if !ok || data["answer"] != answer || data["explanation"] != explanation { + t.Fatalf("structured result data = %#v", part.Data()) + } + if part.MediaType != "application/json" { + t.Fatalf("structured result media type = %q", part.MediaType) + } + if got, ok := kagenta2a.StructuredOutputSchemaSHA256(part); !ok || len(got) != 64 { + t.Fatalf("structured result schema digest = %#v", got) + } +} + func TestOpaqueBYOAgentInteraction(t *testing.T) { fixture := newInteractionFixtureForHarnessTemplate(t, interactionTarget(t), "byo-e2e", "byo-smoke") for range 2 { @@ -841,6 +894,44 @@ func createInteractionTemplate(t *testing.T, modelURL string) string { return template.Name } +func createStructuredOutputInteractionTemplate(t *testing.T, kube ctrlclient.Client, modelURL, mcpURL string) *v1alpha3.AgentTemplate { + t.Helper() + model := createInteractionModel(t, kube, modelURL, nil) + server := &v1alpha3.RemoteMCPServer{ + ObjectMeta: metav1.ObjectMeta{GenerateName: "structured-output-mcp-", Namespace: "kagent"}, + Spec: v1alpha3.RemoteMCPServerSpec{ + Description: "Structured output interaction E2E fixture", + Protocol: v1alpha3.RemoteMCPServerProtocolStreamableHttp, + URL: mcpURL, + }, + } + if err := kube.Create(t.Context(), server); err != nil { + t.Fatalf("create structured-output RemoteMCPServer: %v", err) + } + t.Cleanup(func() { + if err := kube.Delete(context.Background(), server); err != nil && !apierrors.IsNotFound(err) { + t.Errorf("delete structured-output RemoteMCPServer: %v", err) + } + }) + template := &v1alpha3.AgentTemplate{ + ObjectMeta: metav1.ObjectMeta{ + GenerateName: "structured-output-", Namespace: "kagent", + Labels: map[string]string{"kagent.dev/e2e-runtime": "kagent", "kagent.dev/harness": "kagent"}, + }, + Spec: v1alpha3.AgentTemplateSpec{ + ModelConfig: &corev1.LocalObjectReference{Name: model.Name}, + SystemPrompt: "Use available tools when needed, then return the arithmetic answer and a short explanation.", + OutputSchema: &apiextensionsv1.JSON{Raw: []byte(structuredOutputSchema)}, + Tools: []v1alpha3.ToolBinding{{MCP: &v1alpha3.MCPToolBinding{ + Server: corev1.TypedLocalObjectReference{Kind: "RemoteMCPServer", Name: server.Name}, + Tools: []string{"add_numbers"}, + }}}, + }, + } + createAndWaitInteractionTemplate(t, kube, template) + return template +} + func createMCPInteractionTemplate(t *testing.T, modelURL, mcpURL string) string { return createMCPInteractionTemplateForHarness(t, modelURL, mcpURL, "kagent", "kagent") } diff --git a/go/core/test/e2e/mocks/invoke_structured_output.json b/go/core/test/e2e/mocks/invoke_structured_output.json new file mode 100644 index 0000000000..25bf99b202 --- /dev/null +++ b/go/core/test/e2e/mocks/invoke_structured_output.json @@ -0,0 +1,67 @@ +{ + "openai": [ + { + "name": "structured_output_tool_call", + "match": { + "match_type": "contains", + "message": { + "content": "Add 3 and 5 and return the structured result.", + "role": "user" + } + }, + "response": { + "id": "chatcmpl-structured-output-tool-1", + "object": "chat.completion", + "created": 1677652288, + "model": "gpt-4.1-mini", + "choices": [ + { + "index": 0, + "message": { + "content": "", + "role": "assistant", + "tool_calls": [ + { + "id": "structured_call_1", + "type": "function", + "function": { + "name": "add_numbers", + "arguments": "{\"a\":3,\"b\":5}" + } + } + ] + }, + "finish_reason": "tool_calls" + } + ] + } + }, + { + "name": "structured_output_after_tool", + "match": { + "match_type": "contains", + "message": { + "content": "8", + "role": "tool", + "tool_call_id": "structured_call_1" + } + }, + "response": { + "id": "chatcmpl-structured-output-tool-2", + "object": "chat.completion", + "created": 1677652288, + "model": "gpt-4.1-mini", + "choices": [ + { + "index": 0, + "message": { + "content": "{\"answer\":8,\"explanation\":\"three plus five equals eight\"}", + "role": "assistant" + }, + "finish_reason": "stop" + } + ] + } + } + ] +} diff --git a/helm/kagent-crds/templates/kagent.dev_agenttemplates.yaml b/helm/kagent-crds/templates/kagent.dev_agenttemplates.yaml index 8738a15588..eb52e4a103 100644 --- a/helm/kagent-crds/templates/kagent.dev_agenttemplates.yaml +++ b/helm/kagent-crds/templates/kagent.dev_agenttemplates.yaml @@ -65,6 +65,27 @@ spec: x-kubernetes-validations: - message: name must not be empty rule: has(self.name) && self.name != '' + outputSchema: + description: |- + OutputSchema constrains successful terminal output when this template is + compiled as the root agent. + type: object + x-kubernetes-preserve-unknown-fields: true + outputSchemaFrom: + description: |- + OutputSchemaFrom references a JSON Schema stored as JSON in a + same-namespace ConfigMap key. + properties: + key: + minLength: 1 + type: string + name: + minLength: 1 + type: string + required: + - key + - name + type: object plugins: items: description: PluginBundle selects Agent Skills from one immutable @@ -394,6 +415,8 @@ spec: x-kubernetes-validations: - message: systemPrompt and systemPromptFrom are mutually exclusive rule: '!(has(self.systemPrompt) && has(self.systemPromptFrom))' + - message: outputSchema and outputSchemaFrom are mutually exclusive + rule: '!(has(self.outputSchema) && has(self.outputSchemaFrom))' status: description: AgentTemplateStatus is the controller-observed state for each admitting Harness. diff --git a/helm/kagent-crds/templates/kagent.dev_harnesses.yaml b/helm/kagent-crds/templates/kagent.dev_harnesses.yaml index dd1868e807..b186ddabda 100644 --- a/helm/kagent-crds/templates/kagent.dev_harnesses.yaml +++ b/helm/kagent-crds/templates/kagent.dev_harnesses.yaml @@ -385,6 +385,11 @@ spec: type: boolean streaming: type: boolean + structuredOutput: + description: |- + StructuredOutput reports whether the pinned adapter can enforce a root + JSON output contract and emit it as an A2A DataPart. + type: boolean version: description: Version identifies the controller capability catalog entry. @@ -401,6 +406,7 @@ spec: - nativeAgentTools - resume - streaming + - structuredOutput - version type: object conditions: diff --git a/ui/src/api/chat/a2aGrpcChatClient.test.ts b/ui/src/api/chat/a2aGrpcChatClient.test.ts index 86c418ceab..5eed93cf79 100644 --- a/ui/src/api/chat/a2aGrpcChatClient.test.ts +++ b/ui/src/api/chat/a2aGrpcChatClient.test.ts @@ -50,8 +50,13 @@ const text = (value: string) => ({ content: { case: "text" as const, value } }); * the point, since the client reading one as if it were JSON is the bug this file * caught. */ -const data = (value: Record) => ({ +const data = ( + value: Record, + options: { mediaType?: string; metadata?: Record } = {}, +) => ({ content: { case: "data" as const, value: fromJson(ValueSchema, value as never) }, + mediaType: options.mediaType ?? "", + metadata: options.metadata, }); /** A `working` status update carrying one message. */ @@ -670,6 +675,74 @@ describe("A2AGrpcChatClient.send", () => { expect(kinds).toEqual(["tool_call", "tool_result"]); }); + it("keeps a streamed JSON result distinct from tool traffic", async () => { + const result = data( + { payload: { customerId: "12345" }, status: "success" }, + { + mediaType: "application/json", + metadata: { "kagent.dev/a2a/output-schema-sha256": "abc123" }, + }, + ); + const events = await turn([ + { + payload: { + case: "artifactUpdate" as const, + value: { + taskId: "task-1", + contextId: CONVERSATION.contextId, + artifact: { artifactId: "answer", parts: [result] }, + lastChunk: true, + }, + }, + }, + statusFrame({ state: TaskState.COMPLETED }), + ]); + + const part = transcript(events) + .flatMap((message) => message.parts) + .find((candidate) => candidate.kind === "data"); + expect(part).toEqual({ + kind: "data", + dataKind: "structured_output", + data: { payload: { customerId: "12345" }, status: "success" }, + mediaType: "application/json", + metadata: { "kagent.dev/a2a/output-schema-sha256": "abc123" }, + }); + }); + + it('keeps structured output whose user data contains name "ask_user"', async () => { + const result = data( + { name: "ask_user", answer: 4 }, + { + mediaType: "application/json", + metadata: { "kagent.dev/a2a/output-schema-sha256": "abc123" }, + }, + ); + const events = await turn([ + { + payload: { + case: "artifactUpdate" as const, + value: { + taskId: "task-1", + contextId: CONVERSATION.contextId, + artifact: { artifactId: "answer", parts: [result] }, + lastChunk: true, + }, + }, + }, + statusFrame({ state: TaskState.COMPLETED }), + ]); + + const structured = transcript(events) + .flatMap((message) => message.parts) + .find( + (part) => part.kind === "data" && part.dataKind === "structured_output", + ); + expect(structured).toMatchObject({ + data: { name: "ask_user", answer: 4 }, + }); + }); + it("reports the turn reaching completion", async () => { const events = await turn([ statusFrame({ @@ -1172,6 +1245,40 @@ describe("A2AGrpcChatClient.history", () => { expect(messages).toHaveLength(1); }); + it("restores a persisted structured result with its contract metadata", async () => { + serveTasks([ + { + id: "task-1", + contextId: CONVERSATION.contextId, + status: { state: TaskState.COMPLETED, timestamp: { seconds: 1767225600n } }, + history: [], + artifacts: [ + { + artifactId: "answer", + parts: [ + data( + { status: "success" }, + { + mediaType: "application/json", + metadata: { "kagent.dev/a2a/output-schema-sha256": "abc123" }, + }, + ), + ], + }, + ], + }, + ]); + + const { messages } = await new A2AGrpcChatClient().history(CONVERSATION); + expect(messages[0]?.parts[0]).toEqual({ + kind: "data", + dataKind: "structured_output", + data: { status: "success" }, + mediaType: "application/json", + metadata: { "kagent.dev/a2a/output-schema-sha256": "abc123" }, + }); + }); + it("coalesces persisted artifact chunks without crossing structured parts", async () => { // `append: true` is projected by the gateway as several parts on one artifact. // Those are transport chunks, not separate prose blocks, so reopening a task diff --git a/ui/src/api/chat/a2aGrpcChatClient.ts b/ui/src/api/chat/a2aGrpcChatClient.ts index aa91cb8ab4..9de178bf01 100644 --- a/ui/src/api/chat/a2aGrpcChatClient.ts +++ b/ui/src/api/chat/a2aGrpcChatClient.ts @@ -111,6 +111,7 @@ const HISTORY_PAGE_LIMIT = 50; * https://github.com/a2aproject/A2A/pull/2129 */ const TIMELINE_POSITION_METADATA_KEY = "kagent.dev/timeline-position"; +const OUTPUT_SCHEMA_SHA256_METADATA_KEY = "kagent.dev/a2a/output-schema-sha256"; /** Ids for the messages the wire did not name. */ let counter = 0; @@ -185,7 +186,19 @@ function toPart(part: A2APart): ChatPart | undefined { return undefined; } const data = value as Record; - return { kind: "data", dataKind: dataKindOf(data), data }; + return { + kind: "data", + dataKind: dataKindOf( + data, + part.mediaType, + part.metadata as Record | undefined, + ), + data, + ...(part.mediaType ? { mediaType: part.mediaType } : {}), + ...(part.metadata + ? { metadata: part.metadata as Record } + : {}), + }; } // A file part, by url or raw bytes. Nothing renders one yet, and inventing a // placeholder would put a broken attachment in a transcript that has none. @@ -195,13 +208,24 @@ function toPart(part: A2APart): ChatPart | undefined { /** * What a data part represents. * - * Read from the payload's own shape rather than from a `kind` the wire does not - * carry: the runtime emits a tool call as `{name, args}` and its result as - * `{name, response}`. Anything else is passed through as `unknown` and rendered - * as structured data, which is honest — it is data, and this build does not know - * what kind. + * A structured terminal answer has an explicit runtime-owned signature: JSON media + * type plus the digest of the schema that was enforced. Tool traffic has no wire + * discriminator, so it is read from the payload shape the runtime emits: + * `{name, args}` for a call and `{name, response}` for its result. Requiring both + * pieces of the output signature avoids relabelling an arbitrary JSON tool payload + * as the agent's final answer. */ -function dataKindOf(data: Record): ChatDataPart["dataKind"] { +function dataKindOf( + data: Record, + mediaType: string, + metadata: Record | undefined, +): ChatDataPart["dataKind"] { + if ( + mediaType.split(";", 1)[0]?.trim().toLowerCase() === "application/json" && + typeof metadata?.[OUTPUT_SCHEMA_SHA256_METADATA_KEY] === "string" + ) { + return "structured_output"; + } if ("args" in data) return "tool_call"; if ("response" in data || "result" in data) return "tool_result"; return "unknown"; @@ -243,7 +267,9 @@ function visibleParts( visible.push(part); continue; } - if (part.data.name === "ask_user") continue; + if (part.dataKind !== "structured_output" && part.data.name === "ask_user") { + continue; + } const control = controlFlowResult(part); if (control === "confirmation_required") continue; diff --git a/ui/src/api/chat/types.ts b/ui/src/api/chat/types.ts index cc87018d96..87e3f1047a 100644 --- a/ui/src/api/chat/types.ts +++ b/ui/src/api/chat/types.ts @@ -21,8 +21,17 @@ export interface ChatTextPart { export interface ChatDataPart { kind: "data"; /** What the payload represents, so a renderer can pick a component. */ - dataKind: "tool_call" | "tool_result" | "tool_not_run" | "unknown"; + dataKind: + | "tool_call" + | "tool_result" + | "tool_not_run" + | "structured_output" + | "unknown"; data: Record; + /** A2A media type, retained so JSON answers are not mistaken for tool payloads. */ + mediaType?: string; + /** Includes the output-schema digest attached to a structured terminal result. */ + metadata?: Record; } /** A structured human decision, rendered as an audit-friendly status card. */ diff --git a/ui/src/api/domain/agentTemplates.ts b/ui/src/api/domain/agentTemplates.ts index af6d60993b..d04f431bdf 100644 --- a/ui/src/api/domain/agentTemplates.ts +++ b/ui/src/api/domain/agentTemplates.ts @@ -129,6 +129,13 @@ export interface AgentTemplateSpec { /** Mutually exclusive with `systemPromptFrom` — the CRD rejects both. */ systemPrompt?: string; systemPromptFrom?: ConfigMapKeyRef; + /** + * The JSON Schema for a successful terminal response from this template when it + * runs as the root agent. Mutually exclusive with `outputSchemaFrom`. + */ + outputSchema?: Record; + /** A same-namespace ConfigMap key containing the output schema as JSON. */ + outputSchemaFrom?: ConfigMapKeyRef; promptTemplate?: AgentTemplatePromptSpec; tools?: ToolBinding[]; skills?: AgentTemplateSkill[]; diff --git a/ui/src/components/agent-template-form/AgentTemplateForm.tsx b/ui/src/components/agent-template-form/AgentTemplateForm.tsx index ecf0226881..f4b55226bb 100644 --- a/ui/src/components/agent-template-form/AgentTemplateForm.tsx +++ b/ui/src/components/agent-template-form/AgentTemplateForm.tsx @@ -24,6 +24,7 @@ import { import { draftProblems, type AgentTemplateDraft, + type OutputSource, } from "./agentTemplateDraft"; const { Text, Paragraph } = Typography; @@ -158,6 +159,11 @@ export function AgentTemplateForm({ return (harnesses.data ?? []).filter((harness) => admitsLabels(harness, labels)); }, [draft.labels, harnesses.data]); + const structuredOutputUnsupportedBy = useMemo( + () => wouldAdmit.filter((harness) => harness.runtime !== "kagent"), + [wouldAdmit], + ); + /* * With one harness on the cluster, a new template is labelled for it without being * asked. @@ -343,6 +349,83 @@ export function AgentTemplateForm({ + + +
+ + set("outputSchemaConfigMap", event.target.value) + } + placeholder={placeholder("ConfigMap name")} + {...readOnlyInput} + /> + set("outputSchemaKey", event.target.value)} + placeholder={placeholder("Key")} + {...readOnlyInput} + /> + + ) : readOnly ? ( + none("Successful final answers are returned as text.") + ) : null} + + {draft.outputSource !== "text" && + structuredOutputUnsupportedBy.length > 0 ? ( + harness.name) + .join(", ")} will report this template as incompatible. The controller remains the source of truth for compatibility.`} + /> + ) : null} + + + {/* Tools — MCP servers */} { expect(spec.modelConfig).toEqual({ name: "gpt" }); expect(spec.systemPrompt).toBe("Be brief."); + expect(spec.outputSchema).toEqual(template.resource.spec.outputSchema); expect(spec.tools).toEqual(template.resource.spec.tools); }); @@ -135,6 +141,39 @@ describe("the agent template draft", () => { expect(toInline).not.toHaveProperty("systemPromptFrom"); }); + it("round-trips each output mode without ever sending both schema sources", () => { + const template = templateWithExtras(); + const inlineDraft = draftFromTemplate(template); + + expect(inlineDraft.outputSource).toBe("inline"); + expect(JSON.parse(inlineDraft.outputSchema)).toEqual( + template.resource.spec.outputSchema, + ); + const inline = specFromDraft(inlineDraft, template.resource.spec); + expect(inline.outputSchema).toEqual(template.resource.spec.outputSchema); + expect(inline).not.toHaveProperty("outputSchemaFrom"); + + inlineDraft.outputSource = "configMap"; + inlineDraft.outputSchemaConfigMap = "schemas"; + inlineDraft.outputSchemaKey = "result.json"; + const fromConfigMap = specFromDraft(inlineDraft, inline); + expect(fromConfigMap.outputSchemaFrom).toEqual({ + name: "schemas", + key: "result.json", + }); + expect(fromConfigMap).not.toHaveProperty("outputSchema"); + + const configMapDraft = draftFromTemplate({ + ...template, + resource: { ...template.resource, spec: fromConfigMap }, + }); + expect(configMapDraft.outputSource).toBe("configMap"); + configMapDraft.outputSource = "text"; + const text = specFromDraft(configMapDraft, fromConfigMap); + expect(text).not.toHaveProperty("outputSchema"); + expect(text).not.toHaveProperty("outputSchemaFrom"); + }); + it("keeps an MCP binding with no selection because it exposes every server tool", () => { const draft = emptyDraft("kagent"); draft.modelConfig = "gpt"; @@ -202,6 +241,36 @@ describe("the agent template draft", () => { expect(draftProblems(draft, { isCreate: true })).toHaveLength(1); }); + it("reports incomplete or malformed structured-output configuration", () => { + const draft = emptyDraft("kagent"); + draft.name = "t"; + draft.modelConfig = "gpt"; + draft.outputSource = "inline"; + + expect(draftProblems(draft, { isCreate: true })).toContain( + "An inline output schema is required.", + ); + + draft.outputSchema = "not json"; + expect(draftProblems(draft, { isCreate: true })).toContain( + "The inline output schema must be a valid JSON object.", + ); + + draft.outputSchema = '{"type":"string"}'; + expect(draftProblems(draft, { isCreate: true })).toContain( + 'The output schema must have "type": "object" at its root.', + ); + + draft.outputSchema = '{"type":"object"}'; + expect(draftProblems(draft, { isCreate: true })).toEqual([]); + + draft.outputSource = "configMap"; + draft.outputSchemaConfigMap = "schemas"; + expect(draftProblems(draft, { isCreate: true })).toContain( + "An output schema read from a ConfigMap needs both the ConfigMap's name and the key inside it.", + ); + }); + it("carries labels through, because they decide whether anything runs it", () => { const draft = emptyDraft("kagent"); draft.labels = [ diff --git a/ui/src/components/agent-template-form/agentTemplateDraft.ts b/ui/src/components/agent-template-form/agentTemplateDraft.ts index abade83650..d1c7e1f894 100644 --- a/ui/src/components/agent-template-form/agentTemplateDraft.ts +++ b/ui/src/components/agent-template-form/agentTemplateDraft.ts @@ -3,8 +3,8 @@ * * ## The property that matters most here * - * **An edit must not delete what the form does not show.** `AgentTemplateSpec` has - * eight fields and this form authors five of them; `skills`, `plugins` and + * **An edit must not delete what the form does not show.** This form authors the + * common interactive fields; `skills`, `plugins` and * `promptTemplate` are rich enough — three artifact-source shapes, each with its own * strict CEL pattern — that authoring them is its own piece of work. * @@ -47,6 +47,9 @@ export interface AgentToolDraft { /** Where the system prompt comes from. The CRD rejects both at once. */ export type PromptSource = "inline" | "configMap"; +/** Whether terminal output is prose or constrained by a JSON Schema. */ +export type OutputSource = "text" | "inline" | "configMap"; + export interface AgentTemplateDraft { name: string; namespace: string; @@ -57,6 +60,11 @@ export interface AgentTemplateDraft { systemPrompt: string; systemPromptConfigMap: string; systemPromptKey: string; + outputSource: OutputSource; + /** Pretty-printed JSON while loaded; kept as text so incomplete edits remain editable. */ + outputSchema: string; + outputSchemaConfigMap: string; + outputSchemaKey: string; mcpTools: McpToolDraft[]; agentTools: AgentToolDraft[]; /** @@ -80,6 +88,10 @@ export function emptyDraft(namespace: string): AgentTemplateDraft { systemPrompt: "", systemPromptConfigMap: "", systemPromptKey: "", + outputSource: "text", + outputSchema: "", + outputSchemaConfigMap: "", + outputSchemaKey: "", mcpTools: [], agentTools: [], labels: [], @@ -103,6 +115,16 @@ export function draftFromTemplate(template: AgentTemplate): AgentTemplateDraft { systemPrompt: spec.systemPrompt ?? "", systemPromptConfigMap: spec.systemPromptFrom?.name ?? "", systemPromptKey: spec.systemPromptFrom?.key ?? "", + outputSource: spec.outputSchemaFrom + ? "configMap" + : spec.outputSchema + ? "inline" + : "text", + outputSchema: spec.outputSchema + ? JSON.stringify(spec.outputSchema, null, 2) + : "", + outputSchemaConfigMap: spec.outputSchemaFrom?.name ?? "", + outputSchemaKey: spec.outputSchemaFrom?.key ?? "", mcpTools: tools .filter((binding) => binding.mcp) .map((binding) => ({ @@ -190,6 +212,29 @@ export function specFromDraft( setOrDelete(spec, "systemPrompt", draft.systemPrompt.trim()); } + /* + * Output sources have the same exactly-one shape as prompt sources, with one + * additional state: ordinary text output means neither schema field is present. + * Parsing happens here only after `draftProblems` has admitted the value. Keeping + * invalid JSON out of the resource also makes this function safe for callers that + * build a preview before enabling Save. + */ + if (draft.outputSource === "configMap") { + delete spec.outputSchema; + const name = draft.outputSchemaConfigMap.trim(); + const key = draft.outputSchemaKey.trim(); + if (name && key) spec.outputSchemaFrom = { name, key }; + else delete spec.outputSchemaFrom; + } else if (draft.outputSource === "inline") { + delete spec.outputSchemaFrom; + const schema = parseOutputSchema(draft.outputSchema); + if (schema) spec.outputSchema = schema; + else delete spec.outputSchema; + } else { + delete spec.outputSchema; + delete spec.outputSchemaFrom; + } + // An empty list is removed rather than sent: `tools: []` and no `tools` mean the // same thing to the controller, and the shorter resource is the one a reader can // read. @@ -240,6 +285,28 @@ export function draftProblems( ); } } + if (draft.outputSource === "configMap") { + const name = draft.outputSchemaConfigMap.trim(); + const key = draft.outputSchemaKey.trim(); + if (name === "" || key === "") { + problems.push( + "An output schema read from a ConfigMap needs both the ConfigMap's name and the key inside it.", + ); + } + } + if (draft.outputSource === "inline") { + const text = draft.outputSchema.trim(); + if (text === "") { + problems.push("An inline output schema is required."); + } else { + const schema = parseOutputSchema(text); + if (!schema) { + problems.push("The inline output schema must be a valid JSON object."); + } else if (schema.type !== "object") { + problems.push('The output schema must have "type": "object" at its root.'); + } + } + } for (const tool of draft.agentTools) { if (tool.name.trim() !== "" && tool.description.trim() === "") { problems.push( @@ -256,6 +323,19 @@ function bareName(ref: string): string { return slash === -1 ? ref : ref.slice(slash + 1); } +/** Parses only the object-valued JSON shape the public API accepts. */ +function parseOutputSchema(text: string): Record | undefined { + try { + const value: unknown = JSON.parse(text); + if (value === null || typeof value !== "object" || Array.isArray(value)) { + return undefined; + } + return value as Record; + } catch { + return undefined; + } +} + /** * Sets a spec field, or removes it when the value is empty. * diff --git a/ui/src/components/chat/ChatMessageItem.test.tsx b/ui/src/components/chat/ChatMessageItem.test.tsx index c29a5abccb..5ab5b1c877 100644 --- a/ui/src/components/chat/ChatMessageItem.test.tsx +++ b/ui/src/components/chat/ChatMessageItem.test.tsx @@ -5,6 +5,33 @@ import { themeFor } from "@/theme/theme"; import { ChatMessageItem } from "./ChatMessageItem"; describe("ChatMessageItem interaction layout", () => { + it("renders a JSON DataPart as the agent's structured result, not a tool result", () => { + render( + + + , + ); + + expect(screen.getByTestId("chat-structured-output")).toBeInTheDocument(); + expect(screen.getByRole("button", { name: "Copy JSON" })).toBeInTheDocument(); + expect(screen.queryByTestId("chat-tool-result")).not.toBeInTheDocument(); + }); + it("gives a short user-carried ask_user record the full notification lane", () => { render( diff --git a/ui/src/components/chat/ChatMessageItem.tsx b/ui/src/components/chat/ChatMessageItem.tsx index 648018f66c..627a29fb2a 100644 --- a/ui/src/components/chat/ChatMessageItem.tsx +++ b/ui/src/components/chat/ChatMessageItem.tsx @@ -3,6 +3,7 @@ import { useTheme } from "@emotion/react"; import { ExtensionSlot } from "@/appExtensions"; import type { ChatMessage } from "@/api"; import { ToolCallCard } from "./ToolCallCard"; +import { StructuredOutputCard } from "./StructuredOutputCard"; import { MarkdownMessage } from "./MarkdownMessage"; import { ToolApprovalRecord } from "./ToolApprovalRecord"; import { AskUserRecord } from "./AskUserRecord"; @@ -114,7 +115,11 @@ export function ChatMessageItem({
) : null ) : part.kind === "data" ? ( - + part.dataKind === "structured_output" ? ( + + ) : ( + + ) ) : part.kind === "tool_approval" ? ( ) : ( diff --git a/ui/src/components/chat/StructuredOutputCard.tsx b/ui/src/components/chat/StructuredOutputCard.tsx new file mode 100644 index 0000000000..9385039e25 --- /dev/null +++ b/ui/src/components/chat/StructuredOutputCard.tsx @@ -0,0 +1,67 @@ +import { Button, Tag, Typography } from "antd"; +import { useTheme } from "@emotion/react"; +import { Copy, FileJson } from "lucide-react"; +import toast from "react-hot-toast"; +import type { ChatDataPart } from "@/api"; +import { copyText } from "@/components/common/copyText"; +import { stableJson } from "./stableJson"; + +const { Text } = Typography; + +/** A schema-constrained terminal answer, distinct from runtime tool traffic. */ +export function StructuredOutputCard({ part }: { part: ChatDataPart }) { + const theme = useTheme(); + const body = stableJson(part.data); + + async function copy(): Promise { + if (await copyText(body)) toast.success("JSON copied"); + else toast.error("Could not copy JSON"); + } + + return ( +
+
+ + Structured result + JSON + +
+ 
+        {body}
+
+
+ ); +} diff --git a/ui/src/pages/AgentTemplateDetailsPage.tsx b/ui/src/pages/AgentTemplateDetailsPage.tsx index a28c4b6591..c85916fc71 100644 --- a/ui/src/pages/AgentTemplateDetailsPage.tsx +++ b/ui/src/pages/AgentTemplateDetailsPage.tsx @@ -37,6 +37,7 @@ import { useInvalidateAgentTemplates, type AgentTemplateHarnessStatus, } from "@/api"; +import { pairRevisionCondition } from "./agentTemplateRevision"; const { Text, Paragraph } = Typography; @@ -288,15 +289,18 @@ export function AgentTemplateDetailsPage() { }, { /* - * The controller's own `Ready` condition for this pair. Its reason is carried - * verbatim — `ActorTemplatePending` and `Ready` are what the cluster says, and - * paraphrasing them would make the message unsearchable against controller logs. + * The earliest failed controller stage for this pair, or Ready when none failed. + * Looking only for Ready hides compiler failures: structured output on a Codex + * or Claude harness, for example, stops at Compatible=False and has no Ready + * condition to display. The controller's reason and message are carried verbatim + * so the UI cannot drift from the compiler's compatibility decision. */ title: "Revision state", key: "state", render: (_, row) => { - const ready = (row.conditions ?? []).find((entry) => entry.type === "Ready"); - if (!ready) { + const conditions = row.conditions ?? []; + const condition = pairRevisionCondition(conditions); + if (!condition) { return ( Not reported @@ -304,11 +308,26 @@ export function AgentTemplateDetailsPage() { ); } return ( - - - {ready.status === "True" ? "Ready" : (ready.reason ?? "Not ready")} + + + {condition.status === "True" + ? "Ready" + : (condition.reason ?? `${condition.type} failed`)} - + {condition.message ? ( + + {condition.message} + + ) : null} + ); }, }, diff --git a/ui/src/pages/agentTemplateRevision.test.ts b/ui/src/pages/agentTemplateRevision.test.ts new file mode 100644 index 0000000000..e7ca36714d --- /dev/null +++ b/ui/src/pages/agentTemplateRevision.test.ts @@ -0,0 +1,32 @@ +import { describe, expect, it } from "vitest"; +import { pairRevisionCondition } from "./agentTemplateRevision"; + +describe("agent template revision conditions", () => { + it("surfaces a compiler compatibility failure even when Ready was never set", () => { + const compatible = { + type: "Compatible", + status: "False", + reason: "UnsupportedConfiguration", + message: 'Harness "codex" does not support structured output', + }; + + expect( + pairRevisionCondition([ + { type: "Accepted", status: "True" }, + { type: "ResolvedRefs", status: "True" }, + compatible, + ]), + ).toBe(compatible); + }); + + it("uses Ready after every earlier stage succeeds", () => { + const ready = { type: "Ready", status: "True", reason: "Ready" }; + expect( + pairRevisionCondition([ + { type: "Accepted", status: "True" }, + { type: "Compatible", status: "True" }, + ready, + ]), + ).toBe(ready); + }); +}); diff --git a/ui/src/pages/agentTemplateRevision.ts b/ui/src/pages/agentTemplateRevision.ts new file mode 100644 index 0000000000..186627646b --- /dev/null +++ b/ui/src/pages/agentTemplateRevision.ts @@ -0,0 +1,16 @@ +import type { AgentTemplateCondition } from "@/api"; + +const STAGES = ["Accepted", "ResolvedRefs", "Compatible", "Ready"] as const; + +/** The condition that best explains whether a template/harness pair can run. */ +export function pairRevisionCondition( + conditions: readonly AgentTemplateCondition[], +): AgentTemplateCondition | undefined { + for (const type of STAGES) { + const failure = conditions.find( + (condition) => condition.type === type && condition.status === "False", + ); + if (failure) return failure; + } + return conditions.find((condition) => condition.type === "Ready"); +}