mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-12 16:15:24 +00:00
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.
This commit is contained in:
1 parent
68ad5953d7
commit
e2ec3370dd
11 files changed
+108
-3
No files matched your search
@@ -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)
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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"},
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -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",
|
||||
|
||||
Reference in new issue
Block a user