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/MISSING.md b/MISSING.md deleted file mode 100644 index e69de29..0000000 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/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 4037fa4..0a29574 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -54,9 +54,9 @@ 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 | +| 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/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 diff --git a/internal/packages/functions/ai/agent.go b/internal/packages/functions/ai/agent.go index 55f165d..38807a2 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,9 @@ 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)"}, + {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, @@ -81,10 +85,13 @@ 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")), + outputSchema: parseOutputSchema(input), } messages := make([]llm.Message, 0, 4) @@ -125,10 +132,13 @@ 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 + outputSchema []workflow.ParameterSchema } // run drives the reasoning loop until a final answer, an error, or the iteration @@ -138,6 +148,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, @@ -160,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) } @@ -173,6 +200,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 @@ -282,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/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) + }) +} 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") + } +} 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) ✗