From e2ec3370dd8148ffd3ff9f2b974dca7bb8dd940c Mon Sep 17 00:00:00 2001 From: thotam <78630736+thotam@users.noreply.github.com> Date: Wed, 24 Jun 2026 13:42:20 +0700 Subject: [PATCH] feat(webhooks): stream provider responses for server-side runs to enable prompt caching (#1273) Server-side webhook agent runs (sync, async, and admin test) used Stream:false, so OpenAI-compatible routers that only cache streaming requests never populated or served their prompt cache. Webhook runs paid full input-token price on every turn even with a stable session and an identical multi-turn prefix, while WS chat (Stream:true) got cache hits from the 3rd turn. Add gateway.webhook_stream (default true, env GOCLAW_WEBHOOK_STREAM) and apply it to all three run sites. ChatStream returns the fully assembled response, so the payload returned to the caller is unchanged. Set to false to restore non-streaming. --- cmd/gateway_http_wiring.go | 1 + cmd/gateway_lifecycle.go | 1 + docs/webhooks.md | 10 +++++++++ internal/config/config_channels.go | 1 + internal/config/config_load.go | 2 ++ internal/config/config_load_test.go | 35 +++++++++++++++++++++++++++++ internal/config/config_system.go | 1 + internal/http/webhooks_llm.go | 11 +++++++-- internal/webhooks/stream.go | 18 +++++++++++++++ internal/webhooks/stream_test.go | 24 ++++++++++++++++++++ internal/webhooks/worker.go | 7 +++++- 11 files changed, 108 insertions(+), 3 deletions(-) create mode 100644 internal/webhooks/stream.go create mode 100644 internal/webhooks/stream_test.go diff --git a/cmd/gateway_http_wiring.go b/cmd/gateway_http_wiring.go index 5de97d6c..eba9dac4 100644 --- a/cmd/gateway_http_wiring.go +++ b/cmd/gateway_http_wiring.go @@ -234,6 +234,7 @@ func (d *gatewayDeps) wireHTTPHandlersOnServer( sharedWebhookLimiter, // K10: shared limiter nil, // lane: nil → internal default (4-slot); configurable in future via cfg webhooks.ResolveTimeoutSec(d.cfg.Gateway.WebhookSyncTimeoutSec), + webhooks.ResolveStream(d.cfg.Gateway.WebhookStream), ) llmH.SetEncKey(webhookEncKey) // K6: decrypt secret at HMAC verify time d.server.SetWebhookLLMHandler(llmH) diff --git a/cmd/gateway_lifecycle.go b/cmd/gateway_lifecycle.go index cca50b9f..fd15dedf 100644 --- a/cmd/gateway_lifecycle.go +++ b/cmd/gateway_lifecycle.go @@ -173,6 +173,7 @@ func (d *gatewayDeps) runLifecycle( WorkerConcurrency: workerConcurrency, PerTenantConcurrency: 4, AsyncAgentTimeout: webhooks.ResolveTimeoutSec(d.cfg.Gateway.WebhookAsyncTimeoutSec), + Stream: webhooks.ResolveStream(d.cfg.Gateway.WebhookStream), }, ) // K6: decrypt raw secret for outbound HMAC signing using the same key as inbound verify. diff --git a/docs/webhooks.md b/docs/webhooks.md index 19e79c9e..443f1ffd 100644 --- a/docs/webhooks.md +++ b/docs/webhooks.md @@ -285,6 +285,16 @@ Both webhook agent-run deadlines default to **600s** and are capped at **3600s** > Sync mode holds the HTTP connection open for the whole run — a value above an upstream proxy/load-balancer read timeout may be cut before the agent finishes. Async mode returns `202` immediately and runs in the background, so a longer deadline is safe. +#### Prompt caching (internal streaming) + +Server-side webhook agent runs (sync, async, and admin test) stream provider responses internally by default. This lets OpenAI-compatible providers/routers populate and serve their prompt cache for the large, stable system+tools+history prefix — non-streaming requests are not cached by some routers, so webhook runs would otherwise pay full input-token price on every turn even with a stable session. The response returned to the caller is unchanged (the gateway assembles the streamed chunks into the same JSON/SSE payload). + +| Setting (config) | Env var | Applies to | Default | +|------------------|---------|-----------|---------| +| `gateway.webhook_stream` | `GOCLAW_WEBHOOK_STREAM` | sync + async + test webhook agent runs | `true` | + +> Set to `false` to restore non-streaming requests (e.g. for a provider that misbehaves when streaming). Caching for `cache_control`-style providers (Anthropic/DashScope) is unaffected by this flag. + ### Async Response — 202 Accepted ```json diff --git a/internal/config/config_channels.go b/internal/config/config_channels.go index d1ddc92f..41c8353f 100644 --- a/internal/config/config_channels.go +++ b/internal/config/config_channels.go @@ -428,6 +428,7 @@ type GatewayConfig struct { TaskRecoveryIntervalSec int `json:"task_recovery_interval_sec,omitempty"` // team task recovery ticker interval in seconds (default 300 = 5min) WebhookAsyncTimeoutSec int `json:"webhook_async_timeout_sec,omitempty"` // async webhook worker agent-run deadline in seconds (default 600, cap 3600) WebhookSyncTimeoutSec int `json:"webhook_sync_timeout_sec,omitempty"` // sync + test webhook handler agent-run deadline in seconds (default 600, cap 3600). NOTE: sync holds the HTTP connection open for this duration — a value above an upstream proxy/LB read timeout may be cut. + WebhookStream *bool `json:"webhook_stream,omitempty"` // stream provider responses for server-side webhook agent runs (sync/async/test) so the upstream can populate/serve its prompt cache (default true). Response to the caller is unchanged (still assembled JSON). BackgroundProvider string `json:"background_provider,omitempty"` // LLM provider for background workers (vault enrichment, consolidation) BackgroundModel string `json:"background_model,omitempty"` // LLM model for background workers PublicURL string `json:"public_url,omitempty"` // public base URL for OAuth callbacks (e.g. "https://goclaw.example.com") diff --git a/internal/config/config_load.go b/internal/config/config_load.go index 5bae3b7c..07429b63 100644 --- a/internal/config/config_load.go +++ b/internal/config/config_load.go @@ -274,6 +274,8 @@ func (c *Config) applyEnvOverrides() { *dst = parseEnvBool(v) } } + // Webhook internal streaming toggle (default true; nil → on via webhooks.ResolveStream). + envBoolPtr("GOCLAW_WEBHOOK_STREAM", &c.Gateway.WebhookStream) envBoolPtr("GOCLAW_SKILLS_SLASH_COMMANDS_ENABLED", &c.Skills.SlashCommands.Enabled) envBoolPtr("GOCLAW_SKILLS_SLASH_COMMANDS_SUGGEST_NOT_FOUND", &c.Skills.SlashCommands.SuggestNotFound) envBool("GOCLAW_SKILLS_SLASH_COMMANDS_PARTIAL_MATCHING", &c.Skills.SlashCommands.PartialMatching) diff --git a/internal/config/config_load_test.go b/internal/config/config_load_test.go index 184e3a08..57df6e61 100644 --- a/internal/config/config_load_test.go +++ b/internal/config/config_load_test.go @@ -150,6 +150,41 @@ func TestLoad_WebhookTimeoutsFromFileAndEnv(t *testing.T) { } } +func TestLoad_WebhookStreamFromFileAndEnv(t *testing.T) { + dir := t.TempDir() + cfgPath := filepath.Join(dir, "config.json5") + + // Unset by default: nil pointer → ResolveStream defaults to true elsewhere. + os.WriteFile(cfgPath, []byte(`{"gateway":{"port":8080}}`), 0644) + cfg, err := Load(cfgPath) + if err != nil { + t.Fatalf("load error: %v", err) + } + if cfg.Gateway.WebhookStream != nil { + t.Fatalf("default webhook_stream: got %v, want nil", *cfg.Gateway.WebhookStream) + } + + // Explicit false in file. + os.WriteFile(cfgPath, []byte(`{"gateway":{"webhook_stream":false}}`), 0644) + cfg, err = Load(cfgPath) + if err != nil { + t.Fatalf("load error: %v", err) + } + if cfg.Gateway.WebhookStream == nil || *cfg.Gateway.WebhookStream != false { + t.Fatalf("file webhook_stream: got %v, want false", cfg.Gateway.WebhookStream) + } + + // Env overrides the file value. + t.Setenv("GOCLAW_WEBHOOK_STREAM", "true") + cfg, err = Load(cfgPath) + if err != nil { + t.Fatalf("load with env error: %v", err) + } + if cfg.Gateway.WebhookStream == nil || *cfg.Gateway.WebhookStream != true { + t.Fatalf("env webhook_stream: got %v, want true", cfg.Gateway.WebhookStream) + } +} + func TestLoad_SkillsMaxUploadSizeFromFileAndEnv(t *testing.T) { dir := t.TempDir() cfgPath := filepath.Join(dir, "config.json5") diff --git a/internal/config/config_system.go b/internal/config/config_system.go index d40be0fc..6f90c5b9 100644 --- a/internal/config/config_system.go +++ b/internal/config/config_system.go @@ -54,6 +54,7 @@ func (c *Config) ApplySystemConfigs(configs map[string]string) { integer("gateway.task_recovery_interval_sec", &c.Gateway.TaskRecoveryIntervalSec) integer("gateway.webhook_async_timeout_sec", &c.Gateway.WebhookAsyncTimeoutSec) integer("gateway.webhook_sync_timeout_sec", &c.Gateway.WebhookSyncTimeoutSec) + boolean("gateway.webhook_stream", &c.Gateway.WebhookStream) // Background workers (vault enrichment, consolidation) str("background.provider", &c.Gateway.BackgroundProvider) diff --git a/internal/http/webhooks_llm.go b/internal/http/webhooks_llm.go index f58f8a14..9ac4451c 100644 --- a/internal/http/webhooks_llm.go +++ b/internal/http/webhooks_llm.go @@ -104,6 +104,11 @@ type WebhookLLMHandler struct { // syncTimeout overrides the webhookLLMTimeout default (600s); set from // gateway.webhook_sync_timeout_sec config (and in tests). 0 → use default. syncTimeout time.Duration + // stream controls whether sync + test webhook agent runs stream provider + // responses so the upstream can populate/serve its prompt cache. The response + // returned to the caller is unchanged (assembled JSON). Set from + // gateway.webhook_stream config via webhooks.ResolveStream (default true). + stream bool } // NewWebhookLLMHandler constructs a WebhookLLMHandler. @@ -115,6 +120,7 @@ func NewWebhookLLMHandler( limiter *webhookLimiter, lane *scheduler.Lane, syncTimeout time.Duration, + stream bool, ) *WebhookLLMHandler { if lane == nil { lane = scheduler.NewLane(webhookLaneName, webhookLaneDefaultConcurrency) @@ -126,6 +132,7 @@ func NewWebhookLLMHandler( limiter: limiter, lane: lane, syncTimeout: syncTimeout, + stream: stream, } } @@ -301,7 +308,7 @@ func (h *WebhookLLMHandler) handleSync( ChatID: webhook.ID.String(), RunID: runID, UserID: req.UserID, - Stream: false, + Stream: h.stream, ModelOverride: req.Model, ExtraSystemPrompt: extraSystemPrompt, HistoryLimit: 0, @@ -562,7 +569,7 @@ func (h *WebhookLLMHandler) RunTest(ctx context.Context, wh *store.WebhookData, Channel: "webhook", ChatID: wh.ID.String(), RunID: runID, - Stream: false, + Stream: h.stream, ModelOverride: model, TraceName: "webhook.llm.test", TraceTags: []string{"webhook", "test"}, diff --git a/internal/webhooks/stream.go b/internal/webhooks/stream.go new file mode 100644 index 00000000..33927b5a --- /dev/null +++ b/internal/webhooks/stream.go @@ -0,0 +1,18 @@ +package webhooks + +// DefaultStream is the fallback for the webhook internal-streaming flag when +// config is unset. Streaming is on by default so the upstream provider can +// populate/serve its prompt cache for server-side webhook agent runs (sync, +// async, and test). Non-streaming requests are not cached by some +// OpenAI-compatible routers, so webhook runs would otherwise always pay full +// input-token price even with a stable session. +const DefaultStream = true + +// ResolveStream converts the optional gateway.webhook_stream config into a bool: +// nil → DefaultStream (true); otherwise the explicit value. +func ResolveStream(v *bool) bool { + if v == nil { + return DefaultStream + } + return *v +} diff --git a/internal/webhooks/stream_test.go b/internal/webhooks/stream_test.go new file mode 100644 index 00000000..55e49511 --- /dev/null +++ b/internal/webhooks/stream_test.go @@ -0,0 +1,24 @@ +package webhooks + +import "testing" + +func TestResolveStream(t *testing.T) { + tr := true + fa := false + cases := []struct { + name string + in *bool + want bool + }{ + {"nil uses default (true)", nil, true}, + {"explicit true", &tr, true}, + {"explicit false", &fa, false}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + if got := ResolveStream(c.in); got != c.want { + t.Fatalf("ResolveStream(%v) = %v, want %v", c.in, got, c.want) + } + }) + } +} diff --git a/internal/webhooks/worker.go b/internal/webhooks/worker.go index bb967dab..9bb2cd6d 100644 --- a/internal/webhooks/worker.go +++ b/internal/webhooks/worker.go @@ -117,6 +117,11 @@ type WorkerConfig struct { // AsyncAgentTimeout is the deadline for each async agent run. 0 → DefaultAgentTimeout (600s). AsyncAgentTimeout time.Duration + + // Stream controls whether the async agent run streams provider responses so the + // upstream can populate/serve its prompt cache. Resolve via webhooks.ResolveStream + // (default true) before constructing the config. + Stream bool } // WebhookWorker is the background callback delivery service. It is started once per @@ -758,7 +763,7 @@ func (w *WebhookWorker) invokeAgent( ChatID: call.WebhookID.String(), RunID: runID, UserID: req.UserID, - Stream: false, + Stream: w.cfg.Stream, ModelOverride: req.Model, ExtraSystemPrompt: extraSystem, TraceName: "webhook.async",