feat(ai): structured output for ai/chat & ai/agent (ADR-0030)

docs/adr/0030-structured-output-enforcement.md

@@ -1,6 +1,6 @@
 # 0030. Structured / JSON output enforcement for `ai/chat` & `ai/agent`
 
-- Status: Proposed
+- Status: Accepted (tool-forced "respond" path shipped; provider-native `response_format` deferred)
 - Date: 2026-06-02
 - Deciders: FUSE maintainers
 
@@ -68,6 +68,17 @@ provider supports native structured output, use it (B); otherwise fall back to t
 
 ## More Information
 
+- **Shipped (Option C, tool-forced — universal)**: `ai/chat` and `ai/agent` accept an optional
+  `outputSchema` input (a list of `{name,type,required,description}` ParameterSchema). When set, the
+  node offers a single forced `respond` tool whose parameters are the JSON Schema
+  (`ParameterSchemaToJSONSchema`) with `ToolChoice: "required"`, parses the tool-call arguments, and
+  validates them against the schema; one bounded repair retry on invalid output, then a
+  `FunctionError`. The validated object becomes the node's `output` (instead of free-form text); the
+  agent runs its normal loop, then coerces the final answer. Absent `outputSchema` = unchanged.
+  Works on every provider (all support tool calling). Logic in
+  `internal/packages/functions/ai/structured.go`. **Deferred (Option B optimization)**:
+  provider-native `response_format` / `json_schema` with a per-provider capability map for providers
+  that support it natively.
 - Reuse: `internal/packages/functions/ai/tools.go` (`ParameterSchemaToJSONSchema`),
   `pkg/workflow/metadata.go` (`ParameterSchema`).
 - Related: [ADR-0006](0006-llm-provider-abstraction-and-multi-provider-strategy.md),

docs/adr/README.md

@@ -56,7 +56,7 @@ copying [`template.md`](template.md).
 | 0027 | [Async tool invocation via a sub-execution channel](0027-async-tool-invocation-sub-execution-channel.md) | Proposed | 2026-06-02 |
 | 0028 | [Agent prompt/context & conversation-memory model](0028-agent-prompt-context-and-memory-model.md) | Accepted | 2026-06-02 |
 | 0029 | [LLM cost & token-usage tracking and budgets](0029-llm-cost-and-usage-tracking-and-budgets.md) | Accepted | 2026-06-02 |
-| 0030 | [Structured/JSON output enforcement for ai nodes](0030-structured-output-enforcement.md) | Proposed | 2026-06-02 |
+| 0030 | [Structured/JSON output enforcement for ai nodes](0030-structured-output-enforcement.md) | Accepted | 2026-06-02 |
 | 0031 | [Settings, secrets & environments: a SecretStore seam](0031-settings-secrets-and-environments.md) | Accepted | 2026-06-02 |
 | 0032 | [Sub-workflow composition: child workflows as first-class instances](0032-sub-workflow-composition.md) | Accepted | 2026-06-03 |
 | 0033 | [Dependency injection & app composition with uber-go/fx](0033-dependency-injection-and-app-composition.md) | Accepted | 2026-06-03 |
@@ -66,8 +66,9 @@ copying [`template.md`](template.md).
 ADRs **0026–0030** are a cohesive **agent-capabilities** series for the next phase of the AI agent
 ([ADR-0005](0005-ai-agents-as-workflow-nodes-phased-roadmap.md)) — orchestrator mode (0026), async
 tool invocation (0027), prompt/context & memory (0028), cost/usage tracking & budgets (0029), and
-structured-output enforcement (0030). The leaf capabilities are being implemented first: **0029**
-Phase A (usage visibility) is `Accepted`/shipped; 0028 and 0030 follow; the larger orchestrator
-(0026) + async-tools (0027) pair stays `Proposed` until reassessed. **0025** (browser-automation
+structured-output enforcement (0030). The leaf capabilities shipped first: **0028**
+(context policy), **0029** Phase A (usage visibility), and **0030** (structured output) are
+`Accepted`. The larger orchestrator (0026) + async-tools (0027) pair stays `Proposed` until
+reassessed. **0025** (browser-automation
 package) is an independent, parallel stream, not part of that series. When an ADR is implemented its
 status moves to `Accepted` and its "More Information" records what shipped (as 0031 does).

internal/packages/functions/ai/agent.go

@@ -45,6 +45,7 @@ func AgentFunctionMetadata() workflow.FunctionMetadata {
 				{Name: "allowedTools", Type: "array", Required: false, Description: "Optional allowlist of full function ids the agent may use as tools; empty means all eligible tools"},
 				{Name: "maxContextTokens", Type: "int", Required: false, Description: "Optional token budget for the running transcript (approximate); 0/absent disables trimming (ADR-0028)"},
 				{Name: "contextStrategy", Type: "string", Required: false, Description: "When over maxContextTokens: 'drop-oldest' (default) or 'summarize' (an extra LLM call summarizes dropped turns)"},
+				{Name: "outputSchema", Type: "array", Required: false, Description: "Optional list of {name,type,required,description} fields; when set the final output is a validated object matching this schema (ADR-0030)"},
 			},
 			Edges: workflow.InputEdgeMetadata{
 				Count:      0,
@@ -90,6 +91,7 @@ func makeAgentFunction(providers llm.Registry, tools ToolRegistry, usage UsageRe
 			usage:            usage,
 			maxContextTokens: input.GetInt("maxContextTokens"),
 			contextStrategy:  contextStrategyOrDefault(input.GetStr("contextStrategy")),
+			outputSchema:     parseOutputSchema(input),
 		}
 
 		messages := make([]llm.Message, 0, 4)
@@ -136,6 +138,7 @@ type agentExecutor struct {
 	usage            UsageRecorder
 	maxContextTokens int
 	contextStrategy  string
+	outputSchema     []workflow.ParameterSchema
 }
 
 // run drives the reasoning loop until a final answer, an error, or the iteration
@@ -173,6 +176,17 @@ func (e *agentExecutor) run(ctx context.Context, messages []llm.Message) workflo
 		messages = append(messages, resp.Message)
 
 		if len(resp.Message.ToolCalls) == 0 {
+			// Structured output (ADR-0030): coerce the final answer into the requested schema.
+			if len(e.outputSchema) > 0 {
+				obj, u, serr := structuredOutput(ctx, e.provider, e.model, messages, e.outputSchema, e.usage, AgentFunctionID)
+				if serr != nil {
+					return errorOutput(fmt.Sprintf("ai/agent: structured output failed: %v", serr))
+				}
+				totalUsage.PromptTokens += u.PromptTokens
+				totalUsage.CompletionTokens += u.CompletionTokens
+				totalUsage.TotalTokens += u.TotalTokens
+				return successOutputData(obj, totalUsage, steps)
+			}
 			return successOutput(resp.Message.Content, totalUsage, steps)
 		}
 
@@ -341,8 +355,14 @@ func marshalToolContent(data map[string]any) string {
 
 // successOutput builds the agent's final successful output.
 func successOutput(answer string, usage llm.Usage, steps []map[string]any) workflow.FunctionOutput {
+	return successOutputData(answer, usage, steps)
+}
+
+// successOutputData builds the agent's final output; output is the answer text or, for structured
+// output (ADR-0030), the validated object.
+func successOutputData(output any, usage llm.Usage, steps []map[string]any) workflow.FunctionOutput {
 	return workflow.NewFunctionSuccessOutput(map[string]any{
-		"output": answer,
+		"output": output,
 		"usage": map[string]any{
 			"promptTokens":     usage.PromptTokens,
 			"completionTokens": usage.CompletionTokens,

internal/packages/functions/ai/chat.go

@@ -57,6 +57,12 @@ func ChatFunctionMetadata() workflow.FunctionMetadata {
 					Required:    false,
 					Description: "Sampling temperature; if omitted the provider default is used",
 				},
+				{
+					Name:        "outputSchema",
+					Type:        "array",
+					Required:    false,
+					Description: "Optional list of {name,type,required,description} fields; when set the output is a validated object matching this schema (ADR-0030)",
+				},
 			},
 			Edges: workflow.InputEdgeMetadata{
 				Count:      0,
@@ -107,6 +113,7 @@ func makeChatFunction(providers llm.Registry, usage UsageRecorder) workflow.Func
 			Messages:    messages,
 			Temperature: optionalTemperature(input),
 		}
+		outputSchema := parseOutputSchema(input)
 
 		// Provider resolution and the completion run in their own goroutine and report back via
 		// Finish so the WorkflowFunc pool worker is freed immediately (mirrors logic/timer).
@@ -123,6 +130,25 @@ func makeChatFunction(providers llm.Registry, usage UsageRecorder) workflow.Func
 				return
 			}
 
+			// Structured output (ADR-0030): coerce the answer into the requested schema.
+			if len(outputSchema) > 0 {
+				obj, u, serr := structuredOutput(ctx, provider, req.Model, messages, outputSchema, usage, ChatFunctionID)
+				if serr != nil {
+					log.Error().Err(serr).Str("provider", provider.Name()).Msg("ai/chat structured output failed")
+					execInfo.Finish(workflow.NewFunctionOutput(workflow.FunctionError, map[string]any{"error": serr.Error()}))
+					return
+				}
+				execInfo.Finish(workflow.NewFunctionSuccessOutput(map[string]any{
+					"output": obj,
+					"usage": map[string]any{
+						"promptTokens":     u.PromptTokens,
+						"completionTokens": u.CompletionTokens,
+						"totalTokens":      u.TotalTokens,
+					},
+				}))
+				return
+			}
+
 			resp, err := provider.Chat(ctx, req)
 			if err != nil {
 				usage.RecordCall(ChatFunctionID, provider.Name(), req.Model, "error")

internal/packages/functions/ai/structured.go

@@ -0,0 +1,147 @@
+package ai
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	"github.com/open-source-cloud/fuse/pkg/llm"
+	"github.com/open-source-cloud/fuse/pkg/workflow"
+)
+
+// respondToolName is the synthetic tool the model is forced to call to deliver structured output.
+const respondToolName = "respond"
+
+// structuredRepairAttempts bounds how many times structuredOutput re-asks on invalid output.
+const structuredRepairAttempts = 2
+
+// parseOutputSchema reads the optional `outputSchema` input (a list of ParameterSchema-shaped
+// objects) into typed schema params. Absent/invalid returns nil (free-form output).
+func parseOutputSchema(input *workflow.FunctionInput) []workflow.ParameterSchema {
+	raw := input.GetAnySliceOrDefault("outputSchema", nil)
+	if len(raw) == 0 {
+		return nil
+	}
+	b, err := json.Marshal(raw)
+	if err != nil {
+		return nil
+	}
+	var schema []workflow.ParameterSchema
+	if json.Unmarshal(b, &schema) != nil {
+		return nil
+	}
+	return schema
+}
+
+// structuredOutput coerces a completion into a typed object matching schema (ADR-0030). It offers a
+// single forced "respond" tool whose parameters are the JSON Schema of the output, parses the tool
+// call's arguments, and validates them; on invalid/malformed output it re-asks once. Tokens are
+// recorded via usage. This tool-forced path works on every provider (all support tool calling);
+// provider-native response_format is a deferred optimization.
+func structuredOutput(ctx context.Context, provider llm.Provider, model string, messages []llm.Message, schema []workflow.ParameterSchema, usage UsageRecorder, function string) (map[string]any, llm.Usage, error) {
+	respondTool := llm.Tool{
+		Name:        respondToolName,
+		Description: "Return the final answer as JSON arguments matching the required schema.",
+		Parameters:  ParameterSchemaToJSONSchema(schema),
+	}
+
+	convo := messages
+	var total llm.Usage
+	var lastErr error
+	for attempt := 0; attempt < structuredRepairAttempts; attempt++ {
+		resp, err := provider.Chat(ctx, llm.ChatRequest{
+			Model:      model,
+			Messages:   convo,
+			Tools:      []llm.Tool{respondTool},
+			ToolChoice: "required",
+		})
+		if err != nil {
+			usage.RecordCall(function, provider.Name(), model, "error")
+			return nil, total, fmt.Errorf("structured output call failed: %w", err)
+		}
+		usage.RecordCall(function, provider.Name(), model, "success")
+		usage.RecordUsage(function, provider.Name(), model, resp.Usage)
+		total.PromptTokens += resp.Usage.PromptTokens
+		total.CompletionTokens += resp.Usage.CompletionTokens
+		total.TotalTokens += resp.Usage.TotalTokens
+
+		obj, perr := parseRespond(resp.Message)
+		if perr != nil {
+			lastErr = perr
+		} else if verr := validateAgainstSchema(obj, schema); verr != nil {
+			lastErr = verr
+		} else {
+			return obj, total, nil
+		}
+		// Repair: re-ask from the ORIGINAL messages plus a note (avoids a dangling tool_call turn).
+		convo = append(append([]llm.Message{}, messages...), llm.Message{
+			Role:    llm.RoleUser,
+			Content: fmt.Sprintf("The previous response was not valid (%v). Call the %q tool again with arguments that match the schema.", lastErr, respondToolName),
+		})
+	}
+	return nil, total, fmt.Errorf("structured output did not match the schema after %d attempts: %w", structuredRepairAttempts, lastErr)
+}
+
+// parseRespond extracts the structured object from a model turn: preferring the `respond` tool
+// call's arguments, falling back to parsing the message content as a JSON object.
+func parseRespond(msg llm.Message) (map[string]any, error) {
+	for _, tc := range msg.ToolCalls {
+		if tc.Name == respondToolName || len(msg.ToolCalls) == 1 {
+			var obj map[string]any
+			if err := json.Unmarshal(tc.Arguments, &obj); err != nil {
+				return nil, fmt.Errorf("tool arguments are not a JSON object: %w", err)
+			}
+			return obj, nil
+		}
+	}
+	if msg.Content != "" {
+		var obj map[string]any
+		if err := json.Unmarshal([]byte(msg.Content), &obj); err == nil {
+			return obj, nil
+		}
+	}
+	return nil, fmt.Errorf("model did not return structured output")
+}
+
+// validateAgainstSchema checks required fields are present and that present fields are of a
+// compatible JSON type. It is intentionally lenient (a guard, not a full JSON-Schema validator).
+func validateAgainstSchema(obj map[string]any, schema []workflow.ParameterSchema) error {
+	for _, p := range schema {
+		v, ok := obj[p.Name]
+		if !ok {
+			if p.Required {
+				return fmt.Errorf("missing required field %q", p.Name)
+			}
+			continue
+		}
+		if !valueMatchesType(v, p.Type) {
+			return fmt.Errorf("field %q has wrong type (want %s)", p.Name, p.Type)
+		}
+	}
+	return nil
+}
+
+// valueMatchesType reports whether a JSON-decoded value is compatible with a FUSE schema type.
+func valueMatchesType(v any, t string) bool {
+	switch t {
+	case "", "any":
+		return true
+	case "string":
+		_, ok := v.(string)
+		return ok
+	case "bool":
+		_, ok := v.(bool)
+		return ok
+	case "int", "float":
+		_, ok := v.(float64) // JSON numbers decode to float64
+		return ok
+	case "array":
+		_, ok := v.([]any)
+		return ok
+	case "map", "object":
+		_, ok := v.(map[string]any)
+		return ok
+	default:
+		return true
+	}
+}