From 8bf9c26eb9d7c7ea099f09dcec915dc3cdfed2d8 Mon Sep 17 00:00:00 2001 From: Gustavo Bertoi Date: Thu, 4 Jun 2026 00:24:28 -0300 Subject: [PATCH 1/4] feat(ai): bounded agent context-assembly policy (ADR-0028 Option A) ai/agent gains optional maxContextTokens (approximate budget) + contextStrategy (drop-oldest default | summarize). Before each completion the running transcript is bounded by trimContext: it preserves the leading system turns + the first user task and the most recent turns, dropping the oldest middle turns; 'summarize' replaces them with one LLM-generated summary turn (opt-in extra call, recorded as a steps entry). Token estimate is a coarse ~4-chars/token heuristic, centralized for a future tokenizer. Absent budget = unchanged behavior. Persisted cross-run session store (Option B) and vector/RAG (Option C) deferred. ADR-0028 Proposed -> Accepted (Option A shipped). Co-Authored-By: Claude Opus 4.8 (1M context) --- ...8-agent-prompt-context-and-memory-model.md | 10 ++- docs/adr/README.md | 2 +- internal/packages/functions/ai/agent.go | 75 ++++++++++++++++-- internal/packages/functions/ai/context.go | 76 ++++++++++++++++++ .../packages/functions/ai/context_test.go | 77 +++++++++++++++++++ 5 files changed, 230 insertions(+), 10 deletions(-) create mode 100644 internal/packages/functions/ai/context.go create mode 100644 internal/packages/functions/ai/context_test.go diff --git a/docs/adr/0028-agent-prompt-context-and-memory-model.md b/docs/adr/0028-agent-prompt-context-and-memory-model.md index 6eeee8e..8eefce4 100644 --- a/docs/adr/0028-agent-prompt-context-and-memory-model.md +++ b/docs/adr/0028-agent-prompt-context-and-memory-model.md @@ -1,6 +1,6 @@ # 0028. Agent prompt / context & conversation-memory model -- Status: Proposed +- Status: Accepted (Option A — bounded context-assembly policy — shipped; session store / RAG deferred) - Date: 2026-06-02 - Deciders: FUSE maintainers @@ -67,6 +67,14 @@ multi-turn requirements are concrete. ## More Information +- **Option A shipped**: `ai/agent` accepts optional `maxContextTokens` (an approximate token budget) + and `contextStrategy` (`drop-oldest` default | `summarize`). Before each completion the running + transcript is bounded by `trimContext` (`internal/packages/functions/ai/context.go`), which + preserves the leading system turns + the first user task and the most recent turns, dropping the + oldest middle turns; `summarize` replaces them with one LLM-generated summary turn (an extra, + opt-in model call — recorded as a `steps` entry). Token estimation is a coarse ~4-chars/token + heuristic, centralized so a real tokenizer can replace it. Absent budget = unchanged behavior. + **Deferred**: Option B (persisted cross-run session store) and Option C (vector/RAG memory). - Current assembly: `internal/packages/functions/ai/agent.go` (per-run message slice). - Related: [ADR-0007](0007-agent-reasoning-loop-and-tools-from-functions.md), [ADR-0019](0019-object-store-payload-externalization.md), diff --git a/docs/adr/README.md b/docs/adr/README.md index 4037fa4..836544d 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -54,7 +54,7 @@ copying [`template.md`](template.md). | 0025 | [Browser automation & web-scraping package](0025-browser-automation-and-web-scraping-package.md) | Proposed | 2026-06-01 | | 0026 | [Agent-as-orchestrator mode](0026-agent-as-orchestrator-mode.md) | Proposed | 2026-06-02 | | 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) | 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 | | 0031 | [Settings, secrets & environments: a SecretStore seam](0031-settings-secrets-and-environments.md) | Accepted | 2026-06-02 | diff --git a/internal/packages/functions/ai/agent.go b/internal/packages/functions/ai/agent.go index 55f165d..ec78d86 100644 --- a/internal/packages/functions/ai/agent.go +++ b/internal/packages/functions/ai/agent.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "strings" "time" "github.com/open-source-cloud/fuse/internal/packages/transport" @@ -42,6 +43,8 @@ func AgentFunctionMetadata() workflow.FunctionMetadata { {Name: "temperature", Type: "float", Required: false, Description: "Sampling temperature; if omitted the provider default is used"}, {Name: "maxIterations", Type: "int", Required: false, Default: defaultMaxIterations, Description: "Maximum reasoning iterations (clamped to [1, 25])"}, {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)"}, }, Edges: workflow.InputEdgeMetadata{ Count: 0, @@ -81,10 +84,12 @@ func makeAgentFunction(providers llm.Registry, tools ToolRegistry, usage UsageRe model: input.GetStr("model"), temp: optionalTemperature(input), maxIters: clampIterations(input.GetInt("maxIterations")), - wfID: execInfo.WorkflowID, - execID: execInfo.ExecID, - environment: execInfo.Environment, - usage: usage, + wfID: execInfo.WorkflowID, + execID: execInfo.ExecID, + environment: execInfo.Environment, + usage: usage, + maxContextTokens: input.GetInt("maxContextTokens"), + contextStrategy: contextStrategyOrDefault(input.GetStr("contextStrategy")), } messages := make([]llm.Message, 0, 4) @@ -125,10 +130,12 @@ type agentExecutor struct { model string temp *float32 maxIters int - wfID workflow.ID - execID workflow.ExecID - environment string - usage UsageRecorder + wfID workflow.ID + execID workflow.ExecID + environment string + usage UsageRecorder + maxContextTokens int + contextStrategy string } // run drives the reasoning loop until a final answer, an error, or the iteration @@ -138,6 +145,12 @@ func (e *agentExecutor) run(ctx context.Context, messages []llm.Message) workflo steps := make([]map[string]any, 0) for i := 0; i < e.maxIters; i++ { + // Bound the growing transcript to the configured token budget (ADR-0028). + if trimmed, step := e.applyContextPolicy(ctx, messages); step != nil { + messages = trimmed + steps = append(steps, step) + } + resp, err := e.provider.Chat(ctx, llm.ChatRequest{ Model: e.model, Messages: messages, @@ -173,6 +186,52 @@ func (e *agentExecutor) run(ctx context.Context, messages []llm.Message) workflo return errorOutput("ai/agent: max iterations reached") } +// applyContextPolicy bounds the running transcript to maxContextTokens (ADR-0028). It returns the +// trimmed messages and a step record, or (nil, nil) when no trimming was needed. +func (e *agentExecutor) applyContextPolicy(ctx context.Context, messages []llm.Message) ([]llm.Message, map[string]any) { + if e.maxContextTokens <= 0 { + return nil, nil + } + kept, dropped := trimContext(messages, e.maxContextTokens) + if len(dropped) == 0 { + return nil, nil + } + step := map[string]any{"context": "trimmed", "strategy": e.contextStrategy, "droppedTurns": len(dropped)} + if e.contextStrategy == contextStrategySummarize { + if summary := e.summarizeDropped(ctx, dropped); summary != "" { + h := headLen(messages) + summaryTurn := llm.Message{Role: llm.RoleSystem, Content: "Summary of earlier turns: " + summary} + out := concatMessages(messages[:h], append([]llm.Message{summaryTurn}, kept[h:]...)) + step["summarized"] = true + return out, step + } + } + return kept, step +} + +// summarizeDropped asks the provider to summarize the dropped turns into one note. Returns "" on +// error, so the caller falls back to plain drop-oldest. +func (e *agentExecutor) summarizeDropped(ctx context.Context, dropped []llm.Message) string { + var b strings.Builder + for _, m := range dropped { + fmt.Fprintf(&b, "%s: %s\n", m.Role, m.Content) + } + resp, err := e.provider.Chat(ctx, llm.ChatRequest{ + Model: e.model, + Messages: []llm.Message{ + {Role: llm.RoleSystem, Content: "Summarize the following conversation turns concisely, preserving facts, tool results, and decisions."}, + {Role: llm.RoleUser, Content: b.String()}, + }, + }) + if err != nil { + log.Warn().Err(err).Msg("ai/agent context summarization failed; dropping oldest turns instead") + return "" + } + e.usage.RecordCall(AgentFunctionID, e.provider.Name(), e.model, "success") + e.usage.RecordUsage(AgentFunctionID, e.provider.Name(), e.model, resp.Usage) + return resp.Message.Content +} + // executeToolCall resolves, invokes, and records a single model-requested tool // call. It never aborts the run: unknown tools, bad arguments, invocation errors, // and tool errors are all fed back to the model as the tool's result so it can diff --git a/internal/packages/functions/ai/context.go b/internal/packages/functions/ai/context.go new file mode 100644 index 0000000..4fdbaee --- /dev/null +++ b/internal/packages/functions/ai/context.go @@ -0,0 +1,76 @@ +package ai + +import "github.com/open-source-cloud/fuse/pkg/llm" + +const ( + // contextStrategyDropOldest drops the oldest middle turns when over budget (deterministic). + contextStrategyDropOldest = "drop-oldest" + // contextStrategySummarize replaces dropped turns with a single LLM-generated summary turn + // (opt-in; introduces an extra model call and nondeterminism — ADR-0028). + contextStrategySummarize = "summarize" + // approxCharsPerToken is the heuristic used by estimateTokens. It is a coarse budget guard, + // not exact billing; centralized so a real tokenizer can replace it later. + approxCharsPerToken = 4 +) + +// contextStrategyOrDefault normalizes the configured strategy, defaulting to drop-oldest. +func contextStrategyOrDefault(s string) string { + if s == contextStrategySummarize { + return contextStrategySummarize + } + return contextStrategyDropOldest +} + +// estimateTokens approximates the token count of a message slice (~4 chars/token), counting both +// text content and tool-call arguments. +func estimateTokens(messages []llm.Message) int { + chars := 0 + for _, m := range messages { + chars += len(m.Content) + for _, tc := range m.ToolCalls { + chars += len(tc.Name) + len(tc.Arguments) + } + } + return chars / approxCharsPerToken +} + +// headLen returns the number of leading messages to always preserve: any leading system +// message(s) plus the first user message (the original task). +func headLen(messages []llm.Message) int { + h := 0 + for h < len(messages) && messages[h].Role == llm.RoleSystem { + h++ + } + if h < len(messages) && messages[h].Role == llm.RoleUser { + h++ + } + return h +} + +// trimContext bounds messages to budget tokens (ADR-0028). It preserves the head (leading system +// turns + first user task) and the most recent turns that fit, returning the oldest middle turns +// as dropped. budget<=0, already-fitting input, or a case where even the recent turns alone exceed +// the budget returns the input unchanged with no dropped turns. +func trimContext(messages []llm.Message, budget int) (kept, dropped []llm.Message) { + if budget <= 0 || estimateTokens(messages) <= budget { + return messages, nil + } + h := headLen(messages) + rest := messages[h:] + start := 0 + for start < len(rest) && estimateTokens(concatMessages(messages[:h], rest[start:])) > budget { + start++ + } + if start == 0 { + return messages, nil + } + return concatMessages(messages[:h], rest[start:]), rest[:start] +} + +// concatMessages returns a fresh slice of a followed by b. +func concatMessages(a, b []llm.Message) []llm.Message { + out := make([]llm.Message, 0, len(a)+len(b)) + out = append(out, a...) + out = append(out, b...) + return out +} diff --git a/internal/packages/functions/ai/context_test.go b/internal/packages/functions/ai/context_test.go new file mode 100644 index 0000000..02fe484 --- /dev/null +++ b/internal/packages/functions/ai/context_test.go @@ -0,0 +1,77 @@ +package ai + +import ( + "strings" + "testing" + + "github.com/open-source-cloud/fuse/pkg/llm" + "github.com/stretchr/testify/assert" +) + +// msg builds a message of n tokens (~n*4 chars) for a role. +func msg(role llm.Role, n int) llm.Message { + return llm.Message{Role: role, Content: strings.Repeat("x", n*approxCharsPerToken)} +} + +func TestContextStrategyOrDefault(t *testing.T) { + t.Parallel() + assert.Equal(t, contextStrategyDropOldest, contextStrategyOrDefault("")) + assert.Equal(t, contextStrategyDropOldest, contextStrategyOrDefault("nonsense")) + assert.Equal(t, contextStrategySummarize, contextStrategyOrDefault("summarize")) +} + +func TestTrimContext(t *testing.T) { + t.Parallel() + + // head = system(10) + user(10) = 20 tokens; then 4 middle turns of 10 each. + build := func() []llm.Message { + return []llm.Message{ + msg(llm.RoleSystem, 10), + msg(llm.RoleUser, 10), + msg(llm.RoleAssistant, 10), + msg(llm.RoleTool, 10), + msg(llm.RoleAssistant, 10), + msg(llm.RoleTool, 10), + } + } + + t.Run("budget disabled returns unchanged", func(t *testing.T) { + t.Parallel() + in := build() + kept, dropped := trimContext(in, 0) + assert.Len(t, kept, len(in)) + assert.Empty(t, dropped) + }) + + t.Run("under budget returns unchanged", func(t *testing.T) { + t.Parallel() + in := build() // 60 tokens + kept, dropped := trimContext(in, 1000) + assert.Len(t, kept, len(in)) + assert.Empty(t, dropped) + }) + + t.Run("over budget drops oldest middle, preserves head + recent", func(t *testing.T) { + t.Parallel() + in := build() // 60 tokens total; head=20 + // budget 40 -> head(20) + at most 2 recent turns (20) fit; drop the 2 oldest middle. + kept, dropped := trimContext(in, 40) + + // head preserved: first two are the original system + user task. + assert.Equal(t, llm.RoleSystem, kept[0].Role) + assert.Equal(t, llm.RoleUser, kept[1].Role) + // the two most recent turns are retained as the tail. + assert.Equal(t, in[len(in)-1], kept[len(kept)-1]) + assert.Equal(t, in[len(in)-2], kept[len(kept)-2]) + assert.Len(t, dropped, 2) + assert.LessOrEqual(t, estimateTokens(kept), 40) + }) + + t.Run("recent turns alone exceeding budget leaves input unchanged", func(t *testing.T) { + t.Parallel() + in := []llm.Message{msg(llm.RoleSystem, 10), msg(llm.RoleUser, 100)} + kept, dropped := trimContext(in, 5) + assert.Len(t, kept, len(in)) + assert.Empty(t, dropped) + }) +} From d20278f473a3c10b7c781ea26ed75cd673a12c81 Mon Sep 17 00:00:00 2001 From: Gustavo Bertoi Date: Thu, 4 Jun 2026 00:33:38 -0300 Subject: [PATCH 2/4] feat(ai): structured output for ai/chat & ai/agent via tool-forced respond (ADR-0030) Opt-in: an outputSchema input (list of {name,type,required,description}) makes the node return a validated object instead of free-form text. A single forced "respond" tool carries the schema (ParameterSchemaToJSONSchema) with ToolChoice=required; the tool-call args are parsed and validated, with one bounded repair retry. Works on every provider (all support tool calling). The agent runs its normal loop then coerces the final answer. Absent outputSchema = unchanged. Provider-native response_format (Option B) deferred. ADR-0030 Proposed -> Accepted. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../adr/0030-structured-output-enforcement.md | 13 +- docs/adr/README.md | 9 +- internal/packages/functions/ai/agent.go | 22 ++- internal/packages/functions/ai/chat.go | 26 ++++ internal/packages/functions/ai/structured.go | 147 ++++++++++++++++++ .../packages/functions/ai/structured_test.go | 104 +++++++++++++ 6 files changed, 315 insertions(+), 6 deletions(-) create mode 100644 internal/packages/functions/ai/structured.go create mode 100644 internal/packages/functions/ai/structured_test.go diff --git a/docs/adr/0030-structured-output-enforcement.md b/docs/adr/0030-structured-output-enforcement.md index cc310d3..f870f48 100644 --- a/docs/adr/0030-structured-output-enforcement.md +++ b/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), diff --git a/docs/adr/README.md b/docs/adr/README.md index 836544d..0a29574 100644 --- a/docs/adr/README.md +++ b/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). diff --git a/internal/packages/functions/ai/agent.go b/internal/packages/functions/ai/agent.go index ec78d86..38807a2 100644 --- a/internal/packages/functions/ai/agent.go +++ b/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, diff --git a/internal/packages/functions/ai/chat.go b/internal/packages/functions/ai/chat.go index 8e8eb03..efecfee 100644 --- a/internal/packages/functions/ai/chat.go +++ b/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") diff --git a/internal/packages/functions/ai/structured.go b/internal/packages/functions/ai/structured.go new file mode 100644 index 0000000..6352b51 --- /dev/null +++ b/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 + } +} diff --git a/internal/packages/functions/ai/structured_test.go b/internal/packages/functions/ai/structured_test.go new file mode 100644 index 0000000..1d9b6bd --- /dev/null +++ b/internal/packages/functions/ai/structured_test.go @@ -0,0 +1,104 @@ +package ai + +import ( + "context" + "encoding/json" + "testing" + "time" + + "github.com/open-source-cloud/fuse/pkg/llm" + "github.com/open-source-cloud/fuse/pkg/workflow" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +var testOutputSchema = []workflow.ParameterSchema{ + {Name: "answer", Type: "string", Required: true}, + {Name: "score", Type: "int", Required: false}, +} + +func TestValidateAgainstSchema(t *testing.T) { + t.Parallel() + assert.NoError(t, validateAgainstSchema(map[string]any{"answer": "hi", "score": float64(5)}, testOutputSchema)) + assert.NoError(t, validateAgainstSchema(map[string]any{"answer": "hi"}, testOutputSchema)) // optional omitted + assert.Error(t, validateAgainstSchema(map[string]any{"score": float64(5)}, testOutputSchema)) // required missing + assert.Error(t, validateAgainstSchema(map[string]any{"answer": 1}, testOutputSchema)) // wrong type +} + +func TestParseRespond(t *testing.T) { + t.Parallel() + + obj, err := parseRespond(llm.Message{ToolCalls: []llm.ToolCall{ + {Name: respondToolName, Arguments: json.RawMessage(`{"answer":"hi"}`)}, + }}) + require.NoError(t, err) + assert.Equal(t, "hi", obj["answer"]) + + obj, err = parseRespond(llm.Message{Content: `{"answer":"yo"}`}) + require.NoError(t, err) + assert.Equal(t, "yo", obj["answer"]) + + _, err = parseRespond(llm.Message{Content: "not json"}) + assert.Error(t, err) +} + +func TestStructuredOutput_ToolForced(t *testing.T) { + t.Parallel() + prov := &stubProvider{ + name: "stub", + resp: llm.ChatResponse{ + Message: llm.Message{ + Role: llm.RoleAssistant, + ToolCalls: []llm.ToolCall{{ID: "1", Name: respondToolName, Arguments: json.RawMessage(`{"answer":"42","score":7}`)}}, + }, + Usage: llm.Usage{PromptTokens: 5, CompletionTokens: 2, TotalTokens: 7}, + }, + } + obj, u, err := structuredOutput(context.Background(), prov, "m", + []llm.Message{{Role: llm.RoleUser, Content: "q"}}, testOutputSchema, NopUsageRecorder{}, ChatFunctionID) + require.NoError(t, err) + assert.Equal(t, "42", obj["answer"]) + assert.Equal(t, float64(7), obj["score"]) + assert.Equal(t, 7, u.TotalTokens) + // The forced request offered exactly the respond tool. + require.Len(t, prov.last.Tools, 1) + assert.Equal(t, respondToolName, prov.last.Tools[0].Name) + assert.Equal(t, "required", prov.last.ToolChoice) +} + +func TestChat_StructuredOutputDeliversObject(t *testing.T) { + prov := &stubProvider{ + name: "stub", + resp: llm.ChatResponse{ + Message: llm.Message{ + Role: llm.RoleAssistant, + ToolCalls: []llm.ToolCall{{ID: "1", Name: respondToolName, Arguments: json.RawMessage(`{"answer":"structured"}`)}}, + }, + }, + } + reg := llm.NewStaticRegistry(map[string]llm.Provider{"stub": prov}, "stub") + + fnInput, err := workflow.NewFunctionInputWith(map[string]any{ + "input": "hello", + "outputSchema": []any{ + map[string]any{"name": "answer", "type": "string", "required": true}, + }, + }) + require.NoError(t, err) + done := make(chan workflow.FunctionOutput, 1) + execInfo := workflow.NewExecutionInfo("wf-1", "exec-1", "", fnInput) + execInfo.Finish = func(out workflow.FunctionOutput) { done <- out } + + res, err := makeChatFunction(reg, NopUsageRecorder{})(execInfo) + require.NoError(t, err) + require.True(t, res.Async) + select { + case out := <-done: + require.Equal(t, workflow.FunctionSuccess, out.Status) + obj, ok := out.Data["output"].(map[string]any) + require.True(t, ok, "output should be a structured object") + assert.Equal(t, "structured", obj["answer"]) + case <-time.After(2 * time.Second): + t.Fatal("timed out") + } +} From 002371b880096f3adacef103c1425e14f2a3c488 Mon Sep 17 00:00:00 2001 From: Gustavo Bertoi Date: Thu, 4 Jun 2026 00:36:43 -0300 Subject: [PATCH 3/4] fix(server): serve Swagger /docs correctly behind a reverse-proxy path prefix When FUSE is hosted under a sub-path (e.g. api.example.com/fuse), the /docs redirect used an absolute /docs/index.html, so the browser dropped the /fuse prefix and 404'd. Now the redirect honors the proxy's X-Forwarded-Prefix header, falling back to a new SERVER_BASE_PATH config; and the Swagger UI loads its spec via a relative URL (doc.json) so it resolves under the prefix automatically (or an absolute base-path URL when SERVER_BASE_PATH is set). Root hosting is unchanged. Co-Authored-By: Claude Opus 4.8 (1M context) --- .env.example | 3 +++ internal/actors/mux_server.go | 19 +++++++++++++++++-- internal/app/config/config.go | 5 +++++ 3 files changed, 25 insertions(+), 2 deletions(-) diff --git a/.env.example b/.env.example index 2a3e535..3387f7a 100644 --- a/.env.example +++ b/.env.example @@ -1,6 +1,9 @@ APP_NAME=Fuse PORT=9090 +# Path prefix when hosted behind a reverse proxy (e.g. api.example.com/fuse) so the Swagger UI at +# /docs works under the prefix. The proxy's X-Forwarded-Prefix header takes precedence if set. +# SERVER_BASE_PATH=/fuse # Database: memory | postgres DB_DRIVER=memory diff --git a/internal/actors/mux_server.go b/internal/actors/mux_server.go index f852ced..04ce1fc 100644 --- a/internal/actors/mux_server.go +++ b/internal/actors/mux_server.go @@ -3,6 +3,7 @@ package actors import ( "net/http" "strconv" + "strings" "ergo.services/ergo/act" "ergo.services/ergo/gen" @@ -84,10 +85,24 @@ func (m *muxServer) Init(_ ...any) error { w.Header().Set("Content-Type", "application/json") _, _ = w.Write([]byte(spec.ReadDoc())) }) + // When FUSE is hosted under a reverse-proxy path prefix (e.g. /fuse), the /docs redirect and + // the Swagger UI spec URL must carry that prefix or the browser drops it. The prefix comes from + // the proxy's X-Forwarded-Prefix header when set, else SERVER_BASE_PATH. + configuredBasePath := strings.TrimRight(m.config.Server.BasePath, "/") muxRouter.HandleFunc("/docs", func(w http.ResponseWriter, r *http.Request) { - http.Redirect(w, r, "/docs/index.html", http.StatusMovedPermanently) + prefix := strings.TrimRight(r.Header.Get("X-Forwarded-Prefix"), "/") + if prefix == "" { + prefix = configuredBasePath + } + http.Redirect(w, r, prefix+"/docs/index.html", http.StatusMovedPermanently) }) - muxRouter.PathPrefix("/docs/").Handler(httpSwagger.WrapHandler) + // The UI loads its spec relatively ("doc.json" resolves under /docs/) so it works + // behind any prefix; a configured base path makes it explicit/absolute. + specURL := "doc.json" + if configuredBasePath != "" { + specURL = configuredBasePath + "/docs/doc.json" + } + muxRouter.PathPrefix("/docs/").Handler(httpSwagger.Handler(httpSwagger.URL(specURL))) m.Log().Info("swagger documentation available at /docs or /docs/index.html") // create and spawn a web server meta-process diff --git a/internal/app/config/config.go b/internal/app/config/config.go index c9313ad..61fd5b5 100644 --- a/internal/app/config/config.go +++ b/internal/app/config/config.go @@ -85,6 +85,11 @@ type ( ServerConfig struct { Host string `env:"HOST" envDefault:"0.0.0.0"` Port string `env:"PORT" envDefault:"9090"` + // BasePath is the URL path prefix the engine is reachable under when behind a + // reverse proxy (e.g. "/fuse" for https://api.example.com/fuse). It is used to build + // the Swagger UI redirect/spec URL so /docs works under the prefix. The proxy's + // X-Forwarded-Prefix header takes precedence when present. Empty = served at root. + BasePath string `env:"SERVER_BASE_PATH"` } // ClusterConfig configuration for ergo distributed clustering From c5d16421b5af78ae590e4c031a9ffb7f95315ac2 Mon Sep 17 00:00:00 2001 From: Gustavo Bertoi Date: Thu, 4 Jun 2026 00:42:20 -0300 Subject: [PATCH 4/4] chore: remove stray MISSING.md (empty) and tests.txt (docker build log) Co-Authored-By: Claude Opus 4.8 (1M context) --- MISSING.md | 0 tests.txt | 277 ----------------------------------------------------- 2 files changed, 277 deletions(-) delete mode 100644 MISSING.md delete mode 100644 tests.txt diff --git a/MISSING.md b/MISSING.md deleted file mode 100644 index e69de29..0000000 diff --git a/tests.txt b/tests.txt deleted file mode 100644 index 1408a2b..0000000 --- a/tests.txt +++ /dev/null @@ -1,277 +0,0 @@ - => => reading from stdin 1.95kB 0.0s - => [fuse-dev-2 internal] load build definition from Dockerfile.dev 0.0s - => => transferring dockerfile: 881B 0.0s - => [fuse-dev-3 internal] load metadata for docker.io/library/golang:1.26-bookworm 1.1s - => [auth] library/golang:pull token for registry-1.docker.io 0.0s - => [fuse-dev-2 internal] load .dockerignore 0.0s - => => transferring context: 69B 0.0s - => [fuse-dev-3 1/4] FROM docker.io/library/golang:1.26-bookworm@sha256:4f4ab2c90005e7e63cb631f0b4427f05422f241622ee3ec4727cc5febbf83e34 0.0s - => [fuse-dev-1 internal] load build context 0.0s - => => transferring context: 55B 0.0s - => CACHED [fuse-dev-1 2/4] WORKDIR /app 0.0s - => CACHED [fuse-dev-1 3/4] COPY go.mod go.sum ./ 0.0s - => CACHED [fuse-dev-1 4/4] RUN apt-get update && apt-get install -y --no-install-recommends ca-certificates curl && rm -rf /var/lib/apt/list 0.0s - => [fuse-dev-3] exporting to image 0.0s - => => exporting layers 0.0s - => => writing image sha256:abd43f8dd2574bc950b74c15fc15535b011baa581c90d8139676a22aee278635 0.0s - => => naming to docker.io/library/fuse-fuse-dev-3 0.0s - => [fuse-dev-1] exporting to image 0.0s - => => exporting layers 0.0s - => => writing image sha256:3658bfa57dfeec2e6c06c991248277211c31adb41aab77005d8b91c8466bccd5 0.0s - => => naming to docker.io/library/fuse-fuse-dev-1 0.0s - => [fuse-dev-2] exporting to image 0.0s - => => exporting layers 0.0s - => => writing image sha256:495906f7ba0d897ca6c5aa00485c93e9a5eae5baa353071b3f39071089fd37ff 0.0s - => => naming to docker.io/library/fuse-fuse-dev-2 0.0s - => [fuse-dev] exporting to image 0.0s - => => exporting layers 0.0s - => => writing image sha256:94fb2feea0c2b67ccf9f85ccd0248a12b7ebfcd190a0fe36eea7c4278126ef31 0.0s - => => naming to docker.io/library/fuse-fuse-dev 0.0s - => [fuse-dev-3] resolving provenance for metadata file 0.0s - => [fuse-dev-2] resolving provenance for metadata file 0.0s - => [fuse-dev-1] resolving provenance for metadata file 0.0s - => [fuse-dev] resolving provenance for metadata file 0.0s -[+] up 9/9 - ✔ Image fuse-fuse-dev-3 Built 1.9s - ✔ Image fuse-fuse-dev Built 1.9s - ✔ Image fuse-fuse-dev-1 Built 1.9s - ✔ Image fuse-fuse-dev-2 Built 1.9s - ✔ Container fuse-postgres Running 0.0s - ✔ Container fuse-dev-3 Started 0.5s - ✔ Container fuse-dev-2 Started 0.5s - ✔ Container fuse-dev-1 Started 0.5s - ✔ Container fuse-dev Started 0.2s -➜ fuse git:(main) ✗ clear - -➜ fuse git:(main) ✗ docker compose -f docker-compose.e2e-ci.yml up --build -d - -WARN[0000] Found orphan containers ([fuse-dev-3 fuse-dev-1 fuse-dev-2 fuse-dev fuse-postgres]) for this project. If you removed or renamed this service in your compose file, you can run this command with the --remove-orphans flag to clean it up. -[+] up 8/8 - ✔ Network fuse_e2e Created 0.0s - ✔ Container fuse-fuse-pg-1 Healthy 3.8s - ✔ Container fuse-fuse-s3-1 Healthy 5.5s - ✔ Container fuse-fuse-pg-init-1 Exited 4.4s - ✔ Container fuse-fuse-migrate-1 Exited 4.9s - ✔ Container fuse-2 Started 5.5s - ✔ Container fuse-3 Started 5.5s - ✔ Container fuse-1 Started 5.5s -direnv: loading ~/open-source-cloud/fuse/.envrc -██████████████████████████████████████████████████ -█▌ ▐█ -█▌ >=======> >=> >=> >=>>=> >=======> ▐█ -█▌ >=> >=> >=> >=> >=> >=> ▐█ -█▌ >=> >=> >=> >=> >=> ▐█ -█▌ >=====> >=> >=> >=> >=====> ▐█ -█▌ >=> >=> >=> >=> >=> ▐█ -█▌ >=> >=> >=> >=> >=> >=> ▐█ -█▌ >=> >====> >=>>=> >=======> ▐█ -█▌ ▐█ -██████████████████████████████████████████████████ - -Welcome to the FUSE Development Environment -Let's build something awesome! -direnv: export +APP_NAME +DB_CONN_MAX_LIFETIME +DB_DRIVER +DB_MAX_IDLE_CONNS +DB_MAX_OPEN_CONNS +DB_POSTGRES_DSN +GROUPID +HA_CLAIM_SWEEP_INTERVAL +HA_ENABLED +HA_HEARTBEAT_INTERVAL +HA_LEASE_TIMEOUT +HA_NODE_ID +LOG_FORMAT +OBJECT_STORE_DRIVER +OBJECT_STORE_FS_BASE_PATH +OBJECT_STORE_KEY_PREFIX +PORT +PROJECT_PATH +USERID -➜ fuse git:(main) ✗ go test -tags=e2e ./tests/e2e -v -count=1 -timeout 10m -=== RUN TestE2E_GET_health -=== PAUSE TestE2E_GET_health -=== RUN TestE2E_GET_v1_packages -=== PAUSE TestE2E_GET_v1_packages -=== RUN TestE2E_GET_v1_packages_byID -=== PAUSE TestE2E_GET_v1_packages_byID -=== RUN TestE2E_PUT_GET_v1_schemas_roundTrip -=== PAUSE TestE2E_PUT_GET_v1_schemas_roundTrip -=== RUN TestE2E_GET_v1_schemas_list -=== PAUSE TestE2E_GET_v1_schemas_list -=== RUN TestE2E_PUT_v1_packages_roundTrip -=== PAUSE TestE2E_PUT_v1_packages_roundTrip -=== RUN TestE2E_POST_v1_workflows_cancel -=== PAUSE TestE2E_POST_v1_workflows_cancel -=== RUN TestE2E_POST_v1_workflows_execs_asyncResult -=== PAUSE TestE2E_POST_v1_workflows_execs_asyncResult -=== RUN TestE2E_POST_v1_awakeables_resolve_notFound -=== PAUSE TestE2E_POST_v1_awakeables_resolve_notFound -=== RUN TestResolveWorkflowsDir_override ---- PASS: TestResolveWorkflowsDir_override (0.00s) -=== RUN TestResolveWorkflowsDir_invalidOverride ---- PASS: TestResolveWorkflowsDir_invalidOverride (0.00s) -=== RUN TestResolveWorkflowsDir_repoRelative ---- PASS: TestResolveWorkflowsDir_repoRelative (0.00s) -=== RUN TestReadSchemaFile_rejectsOutsideDir ---- PASS: TestReadSchemaFile_rejectsOutsideDir (0.00s) -=== RUN TestE2E_TriggerWorkflow_Idempotency_Deduplication -=== PAUSE TestE2E_TriggerWorkflow_Idempotency_Deduplication -=== RUN TestE2E_TriggerWorkflow_Idempotency_DifferentKeys -=== PAUSE TestE2E_TriggerWorkflow_Idempotency_DifferentKeys -=== RUN TestE2E_TriggerWorkflow_NoIdempotencyKey -=== PAUSE TestE2E_TriggerWorkflow_NoIdempotencyKey -=== RUN TestE2E_GET_workflow_trace -=== PAUSE TestE2E_GET_workflow_trace -=== RUN TestE2E_GET_workflow_trace_notFound -=== PAUSE TestE2E_GET_workflow_trace_notFound -=== RUN TestE2E_GET_schema_traces -=== PAUSE TestE2E_GET_schema_traces -=== RUN TestE2E_GET_schema_traces_empty -=== PAUSE TestE2E_GET_schema_traces_empty -=== RUN TestE2E_POST_webhook_notFound -=== PAUSE TestE2E_POST_webhook_notFound -=== RUN TestE2E_POST_webhook_triggersWorkflow -=== PAUSE TestE2E_POST_webhook_triggersWorkflow -=== RUN TestWorkflowExecutionSuite -=== PAUSE TestWorkflowExecutionSuite -=== RUN TestWorkflowPersistenceSuite -=== PAUSE TestWorkflowPersistenceSuite -=== RUN TestWorkflowResilienceSuite -=== PAUSE TestWorkflowResilienceSuite -=== CONT TestE2E_GET_health -=== CONT TestE2E_TriggerWorkflow_NoIdempotencyKey -=== CONT TestWorkflowResilienceSuite -=== CONT TestE2E_POST_v1_workflows_cancel -=== CONT TestE2E_POST_v1_awakeables_resolve_notFound -=== CONT TestE2E_TriggerWorkflow_Idempotency_Deduplication -=== CONT TestE2E_TriggerWorkflow_Idempotency_DifferentKeys -=== CONT TestE2E_PUT_GET_v1_schemas_roundTrip ---- PASS: TestE2E_POST_v1_workflows_cancel (0.01s) -=== CONT TestE2E_PUT_v1_packages_roundTrip ---- PASS: TestE2E_GET_health (0.01s) -=== CONT TestE2E_GET_v1_schemas_list ---- PASS: TestE2E_POST_v1_awakeables_resolve_notFound (0.02s) -=== CONT TestE2E_GET_v1_packages_byID ---- PASS: TestE2E_GET_v1_schemas_list (0.01s) -=== CONT TestE2E_GET_v1_packages ---- PASS: TestE2E_GET_v1_packages (0.04s) -=== CONT TestE2E_POST_v1_workflows_execs_asyncResult ---- PASS: TestE2E_POST_v1_workflows_execs_asyncResult (0.01s) -=== CONT TestE2E_POST_webhook_notFound ---- PASS: TestE2E_GET_v1_packages_byID (0.05s) -=== CONT TestWorkflowPersistenceSuite ---- PASS: TestE2E_PUT_GET_v1_schemas_roundTrip (0.07s) -=== CONT TestWorkflowExecutionSuite ---- PASS: TestE2E_POST_webhook_notFound (0.01s) -=== CONT TestE2E_POST_webhook_triggersWorkflow ---- PASS: TestE2E_TriggerWorkflow_Idempotency_DifferentKeys (0.09s) -=== CONT TestE2E_GET_schema_traces ---- PASS: TestE2E_PUT_v1_packages_roundTrip (0.08s) -=== CONT TestE2E_GET_schema_traces_empty ---- PASS: TestE2E_GET_schema_traces_empty (0.00s) -=== CONT TestE2E_GET_workflow_trace_notFound ---- PASS: TestE2E_GET_workflow_trace_notFound (0.00s) -=== CONT TestE2E_GET_workflow_trace ---- PASS: TestE2E_TriggerWorkflow_Idempotency_Deduplication (0.10s) ---- PASS: TestE2E_TriggerWorkflow_NoIdempotencyKey (0.11s) -=== RUN TestWorkflowPersistenceSuite/TestMultipleWorkflows_IndependentState -=== RUN TestWorkflowResilienceSuite/TestErrorEdge_FollowsRecoveryPath ---- PASS: TestE2E_POST_webhook_triggersWorkflow (0.06s) -=== RUN TestWorkflowExecutionSuite/TestConditionalBranch_Finishes -=== RUN TestWorkflowResilienceSuite/TestParallelRetry_CompletesWithConcurrentRetries -=== NAME TestWorkflowPersistenceSuite/TestMultipleWorkflows_IndependentState - workflow_persistence_suite_e2e_test.go:72: - Error Trace: /Users/gustavo.bertoi/open-source-cloud/fuse/tests/e2e/workflow_persistence_suite_e2e_test.go:72 - Error: Received unexpected error: - workflow 019d7d49-2c62-77ad-af02-592a3853b3ad still in status "untriggered" after 15s - Test: TestWorkflowPersistenceSuite/TestMultipleWorkflows_IndependentState -=== RUN TestWorkflowPersistenceSuite/TestPackageRegistration_PersistsInDB -=== RUN TestWorkflowPersistenceSuite/TestSchemaRoundTrip_PersistsInDB -=== RUN TestWorkflowPersistenceSuite/TestTriggerAndFinish_PersistsState -=== NAME TestWorkflowExecutionSuite/TestConditionalBranch_Finishes - workflow_execution_suite_e2e_test.go:67: - Error Trace: /Users/gustavo.bertoi/open-source-cloud/fuse/tests/e2e/workflow_execution_suite_e2e_test.go:50 - /Users/gustavo.bertoi/open-source-cloud/fuse/tests/e2e/workflow_execution_suite_e2e_test.go:67 - /Users/gustavo.bertoi/.gvm/gos/go1.26/src/runtime/asm_arm64.s:1447 - Error: Received unexpected error: - workflow 019d7d49-2c99-7bd6-b190-3769e77a68b7 still in status "untriggered" after 15s - Test: TestWorkflowExecutionSuite/TestConditionalBranch_Finishes - Messages: workflow small-cond-test should reach terminal state -=== RUN TestWorkflowExecutionSuite/TestDurableExecution_Finishes -=== NAME TestWorkflowResilienceSuite/TestParallelRetry_CompletesWithConcurrentRetries - workflow_resilience_suite_e2e_test.go:75: - Error Trace: /Users/gustavo.bertoi/open-source-cloud/fuse/tests/e2e/workflow_resilience_suite_e2e_test.go:75 - Error: Received unexpected error: - workflow 019d7d49-2d39-7c35-a1fe-f7abfe26be6c still in status "untriggered" after 15s - Test: TestWorkflowResilienceSuite/TestParallelRetry_CompletesWithConcurrentRetries - Messages: workflow should reach terminal state -=== RUN TestWorkflowResilienceSuite/TestRetry_CompletesAfterTransientFailures -=== RUN TestWorkflowResilienceSuite/TestTimeout_FollowsRecoveryPath ---- FAIL: TestWorkflowResilienceSuite (17.06s) - --- PASS: TestWorkflowResilienceSuite/TestErrorEdge_FollowsRecoveryPath (0.21s) - --- FAIL: TestWorkflowResilienceSuite/TestParallelRetry_CompletesWithConcurrentRetries (15.07s) - --- PASS: TestWorkflowResilienceSuite/TestRetry_CompletesAfterTransientFailures (0.41s) - --- PASS: TestWorkflowResilienceSuite/TestTimeout_FollowsRecoveryPath (1.23s) -=== NAME TestE2E_GET_workflow_trace - trace_api_e2e_test.go:39: - Error Trace: /Users/gustavo.bertoi/open-source-cloud/fuse/tests/e2e/trace_api_e2e_test.go:39 - Error: Received unexpected error: - workflow 019d7d49-2c54-7d01-844a-708c44a9b0ad still in status "untriggered" after 30s - Test: TestE2E_GET_workflow_trace ---- FAIL: TestE2E_GET_workflow_trace (30.23s) -=== NAME TestWorkflowPersistenceSuite/TestTriggerAndFinish_PersistsState - workflow_persistence_suite_e2e_test.go:49: - Error Trace: /Users/gustavo.bertoi/open-source-cloud/fuse/tests/e2e/workflow_persistence_suite_e2e_test.go:49 - Error: Received unexpected error: - workflow 019d7d49-671b-7775-8e02-6ed6e5f5f36a still in status "untriggered" after 15s - Test: TestWorkflowPersistenceSuite/TestTriggerAndFinish_PersistsState - Messages: workflow should reach terminal state ---- FAIL: TestWorkflowPersistenceSuite (30.25s) - --- FAIL: TestWorkflowPersistenceSuite/TestMultipleWorkflows_IndependentState (15.01s) - --- PASS: TestWorkflowPersistenceSuite/TestPackageRegistration_PersistsInDB (0.01s) - --- PASS: TestWorkflowPersistenceSuite/TestSchemaRoundTrip_PersistsInDB (0.01s) - --- FAIL: TestWorkflowPersistenceSuite/TestTriggerAndFinish_PersistsState (15.15s) -=== NAME TestE2E_GET_schema_traces - trace_api_e2e_test.go:109: - Error Trace: /Users/gustavo.bertoi/open-source-cloud/fuse/tests/e2e/trace_api_e2e_test.go:109 - Error: "0" is not greater than or equal to "1" - Test: TestE2E_GET_schema_traces ---- FAIL: TestE2E_GET_schema_traces (30.23s) -=== NAME TestWorkflowExecutionSuite/TestDurableExecution_Finishes - workflow_execution_suite_e2e_test.go:77: - Error Trace: /Users/gustavo.bertoi/open-source-cloud/fuse/tests/e2e/workflow_execution_suite_e2e_test.go:50 - /Users/gustavo.bertoi/open-source-cloud/fuse/tests/e2e/workflow_execution_suite_e2e_test.go:77 - /Users/gustavo.bertoi/.gvm/gos/go1.26/src/runtime/asm_arm64.s:1447 - Error: Received unexpected error: - workflow 019d7d49-6747-7a0e-afab-63b56a94f4a1 still in status "untriggered" after 15s - Test: TestWorkflowExecutionSuite/TestDurableExecution_Finishes - Messages: workflow durable-test should reach terminal state -=== RUN TestWorkflowExecutionSuite/TestExpressionCondition_Finishes - workflow_execution_suite_e2e_test.go:72: - Error Trace: /Users/gustavo.bertoi/open-source-cloud/fuse/tests/e2e/workflow_execution_suite_e2e_test.go:50 - /Users/gustavo.bertoi/open-source-cloud/fuse/tests/e2e/workflow_execution_suite_e2e_test.go:72 - /Users/gustavo.bertoi/.gvm/gos/go1.26/src/runtime/asm_arm64.s:1447 - Error: Received unexpected error: - workflow 019d7d49-a253-7a1f-888d-525794b5a019 still in status "untriggered" after 15s - Test: TestWorkflowExecutionSuite/TestExpressionCondition_Finishes - Messages: workflow expression-condition-test should reach terminal state -=== RUN TestWorkflowExecutionSuite/TestParallelFanOut_Finishes - workflow_execution_suite_e2e_test.go:62: - Error Trace: /Users/gustavo.bertoi/open-source-cloud/fuse/tests/e2e/workflow_execution_suite_e2e_test.go:50 - /Users/gustavo.bertoi/open-source-cloud/fuse/tests/e2e/workflow_execution_suite_e2e_test.go:62 - /Users/gustavo.bertoi/.gvm/gos/go1.26/src/runtime/asm_arm64.s:1447 - Error: Received unexpected error: - workflow 019d7d49-ddab-7597-b196-842605623f1d still in status "untriggered" after 15s - Test: TestWorkflowExecutionSuite/TestParallelFanOut_Finishes - Messages: workflow small-test should reach terminal state -=== RUN TestWorkflowExecutionSuite/TestSmallestPipeline_Finishes - workflow_execution_suite_e2e_test.go:57: - Error Trace: /Users/gustavo.bertoi/open-source-cloud/fuse/tests/e2e/workflow_execution_suite_e2e_test.go:50 - /Users/gustavo.bertoi/open-source-cloud/fuse/tests/e2e/workflow_execution_suite_e2e_test.go:57 - /Users/gustavo.bertoi/.gvm/gos/go1.26/src/runtime/asm_arm64.s:1447 - Error: Received unexpected error: - workflow 019d7d4a-189b-7a46-8110-3cec1cabeb5d still in status "untriggered" after 15s - Test: TestWorkflowExecutionSuite/TestSmallestPipeline_Finishes - Messages: workflow smallest-test should reach terminal state -=== RUN TestWorkflowExecutionSuite/TestSumRandBranch_Finishes - workflow_execution_suite_e2e_test.go:87: - Error Trace: /Users/gustavo.bertoi/open-source-cloud/fuse/tests/e2e/workflow_execution_suite_e2e_test.go:87 - Error: Received unexpected error: - workflow 019d7d4a-53bd-7102-90d1-49aa741833fd still in status "untriggered" after 1m0s - Test: TestWorkflowExecutionSuite/TestSumRandBranch_Finishes - Messages: workflow should reach terminal state ---- FAIL: TestWorkflowExecutionSuite (135.71s) - --- FAIL: TestWorkflowExecutionSuite/TestConditionalBranch_Finishes (15.02s) - --- FAIL: TestWorkflowExecutionSuite/TestDurableExecution_Finishes (15.11s) - --- FAIL: TestWorkflowExecutionSuite/TestExpressionCondition_Finishes (15.19s) - --- FAIL: TestWorkflowExecutionSuite/TestParallelFanOut_Finishes (15.09s) - --- FAIL: TestWorkflowExecutionSuite/TestSmallestPipeline_Finishes (15.14s) - --- FAIL: TestWorkflowExecutionSuite/TestSumRandBranch_Finishes (60.04s) -FAIL -FAIL github.com/open-source-cloud/fuse/tests/e2e 136.292s -FAIL -➜ fuse git:(main) ✗