Merge pull request #1467 from itsddvn/fix/audio-provider-resolution-timeout

feat(runtime): strengthen hooks, channels, embeddings, and audio
This commit is contained in:
Clark Cant authored and GitHub committed 2026-07-22 21:00:37 +07:00
commit 696efa28db
44 files changed
+1893 -346

No files matched your search

+1 -1
View File
@@ -851,7 +851,7 @@ func runGateway() {
instanceLoader.SetUsageCapService(usageCapSvc)
instanceLoader.RegisterFactory(channels.TypeTelegram, telegram.FactoryWithStoresAndAudio(pgStores.Agents, pgStores.ConfigPermissions, pgStores.Teams, pgStores.SubagentTasks, pgStores.PendingMessages, audioMgr))
instanceLoader.RegisterFactory(channels.TypeDiscord, discord.FactoryWithStoresAndAudio(pgStores.Agents, pgStores.ConfigPermissions, pgStores.PendingMessages, audioMgr))
instanceLoader.RegisterFactory(channels.TypeFeishu, feishu.FactoryWithPendingStoreAndAudio(pgStores.PendingMessages, audioMgr))
instanceLoader.RegisterFactory(channels.TypeFeishu, feishu.FactoryWithStoresAndAudio(pgStores.Agents, pgStores.ConfigPermissions, pgStores.PendingMessages, audioMgr))
instanceLoader.RegisterFactory(channels.TypeZaloOA, zalo.Factory)
instanceLoader.RegisterFactory(channels.TypeZaloPersonal, zalopersonal.FactoryWithPendingStore(pgStores.PendingMessages))
instanceLoader.RegisterFactory(channels.TypeWhatsApp, whatsapp.FactoryWithDBAudio(pgStores.DB, pgStores.PendingMessages, "pgx", audioMgr, pgStores.BuiltinTools))
+4 -2
View File
@@ -52,8 +52,10 @@ func resolveEmbeddingProvider(
}
}
// 2. Auto-detect: scan DB providers for first with settings.embedding.enabled
allProviders, err := providerStore.ListAllProviders(context.Background())
// 2. Auto-detect only from the master tenant. The selected provider receives
// cross-tenant semantic content during startup maintenance, so a tenant-owned
// endpoint must never be selected for this process-wide responsibility.
allProviders, err := providerStore.ListProviders(masterCtx)
if err != nil {
slog.Warn("failed to list providers for embedding auto-detect", "error", err)
return nil
+68 -1
View File
@@ -98,7 +98,12 @@ func captureEmbeddingRequest(t *testing.T, es *store.EmbeddingSettings) map[stri
t.Fatalf("Decode() error = %v", err)
}
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"data":[{"embedding":[0.1,0.2]}]}`))
_ = json.NewEncoder(w).Encode(map[string]any{
"data": []map[string]any{{
"embedding": make([]float32, store.RequiredMemoryEmbeddingDimensions),
"index": 0,
}},
})
}))
defer server.Close()
@@ -137,3 +142,65 @@ func TestBuildEmbeddingProviderIgnoresIncompatibleStoredDimensions(t *testing.T)
t.Fatalf("dimensions = %v, want fallback 1536", got)
}
}
type embeddingProviderStoreStub struct {
masterProviders []store.LLMProviderData
allProviders []store.LLMProviderData
listTenant uuid.UUID
listAllCalled bool
}
func (s *embeddingProviderStoreStub) CreateProvider(context.Context, *store.LLMProviderData) error {
return nil
}
func (s *embeddingProviderStoreStub) GetProvider(context.Context, uuid.UUID) (*store.LLMProviderData, error) {
return nil, nil
}
func (s *embeddingProviderStoreStub) GetProviderByName(context.Context, string) (*store.LLMProviderData, error) {
return nil, nil
}
func (s *embeddingProviderStoreStub) ListProviders(ctx context.Context) ([]store.LLMProviderData, error) {
s.listTenant = store.TenantIDFromContext(ctx)
return s.masterProviders, nil
}
func (s *embeddingProviderStoreStub) ListAllProviders(context.Context) ([]store.LLMProviderData, error) {
s.listAllCalled = true
return s.allProviders, nil
}
func (s *embeddingProviderStoreStub) UpdateProvider(context.Context, uuid.UUID, map[string]any) error {
return nil
}
func (s *embeddingProviderStoreStub) DeleteProvider(context.Context, uuid.UUID) error { return nil }
func TestResolveEmbeddingProviderAutoDetectUsesMasterTenantOnly(t *testing.T) {
embeddingSettings := json.RawMessage(`{"embedding":{"enabled":true}}`)
providerStore := &embeddingProviderStoreStub{
masterProviders: []store.LLMProviderData{{
TenantID: store.MasterTenantID,
Name: "master-embedding",
ProviderType: store.ProviderOpenAICompat,
APIBase: "http://master.invalid/v1",
APIKey: "master-key",
Enabled: true,
Settings: embeddingSettings,
}},
allProviders: []store.LLMProviderData{{
TenantID: uuid.New(),
Name: "tenant-controlled-endpoint",
ProviderType: store.ProviderOpenAICompat,
Enabled: true,
Settings: embeddingSettings,
}},
}
provider := resolveEmbeddingProvider(providerStore, nil, nil)
if provider == nil || provider.Name() != "master-embedding" {
t.Fatalf("resolveEmbeddingProvider() = %v, want master-embedding", provider)
}
if providerStore.listTenant != store.MasterTenantID {
t.Fatalf("ListProviders tenant = %s, want master tenant", providerStore.listTenant)
}
if providerStore.listAllCalled {
t.Fatal("ListAllProviders was called; cross-tenant auto-detect must stay disabled")
}
}
+45
View File
@@ -420,16 +420,61 @@ func setupMemoryEmbeddings(
}
// Wire embedding provider into vault store for semantic document search.
var vaultStore *pg.PGVaultStore
if pgStores.Vault != nil {
pgStores.Vault.SetEmbeddingProvider(embProvider)
slog.Info("vault embeddings enabled", "provider", embProvider.Name())
vaultStore, _ = pgStores.Vault.(*pg.PGVaultStore)
}
// V3: Wire embedding provider into episodic store for semantic search.
var episodicStore *pg.PGEpisodicStore
if pgStores.Episodic != nil {
pgStores.Episodic.SetEmbeddingProvider(embProvider)
slog.Info("episodic embeddings enabled", "provider", embProvider.Name())
episodicStore, _ = pgStores.Episodic.(*pg.PGEpisodicStore)
}
// Agent create/update embedding hooks require the provider to be wired.
var agentStore *pg.PGAgentStore
if pgAgentStore, ok := pgStores.Agents.(*pg.PGAgentStore); ok {
agentStore = pgAgentStore
agentStore.SetEmbeddingProvider(embProvider)
slog.Info("agent embeddings enabled", "provider", embProvider.Name())
}
// Recover the remaining semantic indexes sequentially to avoid a burst
// of concurrent batch requests during gateway startup. Each surface gets
// its own deadline so a large agent backlog cannot starve later stores.
go func() {
if agentStore != nil {
bgCtx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
if count, err := agentStore.BackfillAgentEmbeddings(bgCtx); err != nil {
slog.Warn("agent embeddings backfill failed", "error", err)
} else if count > 0 {
slog.Info("agent embeddings recovery complete", "agents_updated", count)
}
cancel()
}
if episodicStore != nil {
bgCtx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
if count, err := episodicStore.BackfillEpisodicEmbeddings(bgCtx); err != nil {
slog.Warn("episodic embeddings backfill failed", "error", err)
} else if count > 0 {
slog.Info("episodic embeddings backfill complete", "summaries_updated", count)
}
cancel()
}
if vaultStore != nil {
bgCtx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
if count, err := vaultStore.BackfillVaultEmbeddings(bgCtx); err != nil {
slog.Warn("vault embeddings backfill failed", "error", err)
} else if count > 0 {
slog.Info("vault embeddings backfill complete", "documents_updated", count)
}
cancel()
}
}()
} else {
slog.Warn("memory embeddings disabled (no API key), chunks stored without vectors")
}
+11 -2
View File
@@ -282,8 +282,8 @@ sequenceDiagram
PG-->>GW: PG stores created
GW->>GW: 5. Start tracing collector
GW->>PG: 6. Register providers from DB
GW->>PG: 7. Wire embedding provider to PGMemoryStore
GW->>PG: 8. Backfill memory embeddings (background)
GW->>PG: 7. Wire embedding provider to vector-enabled PG stores
GW->>PG: 8. Backfill missing embeddings in bounded background batches
GW->>GW: 9. Register config-based providers
GW->>GW: 10. Create tool registry (filesystem, exec, web, memory, browser, TTS, subagent, MCP)
@@ -303,6 +303,15 @@ sequenceDiagram
GW->>Engine: 22. Listen on host:port
```
Embedding recovery covers memory chunks, knowledge graph entities, active
skills, non-cancelled team tasks, active agents, episodic summaries, and vault
documents. The process-wide provider is resolved from the master tenant because
startup recovery may process semantic content from multiple tenants. Agent,
episodic, and vault backfills run sequentially with independent deadlines and
bounded batches. They only select rows whose embedding is `NULL`, re-check the
source fields before writing, isolate malformed rows, and can retry remaining
rows on the next startup.
---
## 7. Database Wiring
+15 -1
View File
@@ -87,6 +87,18 @@ Hooks resolve in priority order, highest first. A single `block` decision short-
```
- Non-JSON 2xx → allow.
### script
Custom ES5.1 scripts may return `additionalContext` with an `allow` decision.
For `user_prompt_submit`, GoClaw appends that text to the current run's extra
system prompt without changing the original user message.
```javascript
function handle(event) {
return { decision: "allow", additionalContext: "Call the required tool before answering." };
}
```
### prompt
```json
@@ -95,7 +107,8 @@ Hooks resolve in priority order, highest first. A single `block` decision short-
"matcher": "^(exec|shell|write_file)$",
"config": {
"prompt_template": "Evaluate safety of this tool call.",
"model": "haiku",
"provider": "openai",
"model": "gpt-4.1-mini",
"max_invocations_per_turn": 5
}
}
@@ -104,6 +117,7 @@ Hooks resolve in priority order, highest first. A single `block` decision short-
Required:
- `prompt_template` — system-level instruction the evaluator receives.
- `matcher` or `if_expr` — runaway-cost guard; prevents firing the LLM on every event.
- `provider` and `model` — required for prompt hooks created or edited in the dashboard; select the pair from the configured provider catalog. Legacy model-only hooks remain supported at runtime.
Safeguards:
- **Structured output**: evaluator MUST call a `decide(decision, reason, injection_detected, updated_input)` tool. Free-text responses fail-closed.
+44 -18
View File
@@ -41,9 +41,9 @@ type Manager struct {
musicProviders map[string]MusicProvider
sfxProviders map[string]SFXProvider
primary string // primary TTS provider
sttChain []string // STT fallback order (Phase 4)
musicChain []string // Music fallback order (Phase 3)
primary string // primary TTS provider
sttChain []string // STT fallback order (Phase 4)
musicChain []string // Music fallback order (Phase 3)
channelSTTOverrides map[string][]string // channel → provider key list (Phase 4)
auto AutoMode
@@ -298,33 +298,59 @@ func (m *Manager) SynthesizeWithFallback(ctx context.Context, text string, opts
// SynthesizeWithFallbackAdapted is like SynthesizeWithFallback but applies
// AdaptAgentParams(genericAgentParams, providerName) per-attempt before
// synthesizing. This is the Finding #1 fix: each fallback attempt receives
// provider-native params rather than the primary's adapted keys.
// synthesizing, so each fallback attempt receives provider-native params
// rather than the primary's adapted keys.
//
// genericAgentParams must use the generic allow-list keys (speed, emotion, style).
// Passing nil is safe and produces the same behaviour as SynthesizeWithFallback.
func (m *Manager) SynthesizeWithFallbackAdapted(ctx context.Context, text string, opts TTSOptions, genericAgentParams map[string]any) (*SynthResult, error) {
var providerErrs []error
if p, ok := m.ttsProviders[m.primary]; ok {
attemptOpts := m.withAdaptedParams(opts, m.primary, genericAgentParams)
if result, err := p.Synthesize(ctx, text, attemptOpts); err == nil {
return result, nil
} else {
slog.Warn("tts primary provider failed, trying fallback", "provider", m.primary, "error", err)
providerErrs = append(providerErrs, fmt.Errorf("%s: %w", m.primary, err))
return m.SynthesizeWithFallbackResolved(ctx, text, m.primary, func(string) TTSOptions {
return opts
}, genericAgentParams)
}
// SynthesizeWithFallbackResolved tries preferredProvider first, then the
// manager primary and remaining registered providers. resolveOpts runs for
// every actual attempt so provider-specific voice, model, and format defaults
// never leak into another provider's request.
func (m *Manager) SynthesizeWithFallbackResolved(
ctx context.Context,
text string,
preferredProvider string,
resolveOpts func(providerName string) TTSOptions,
genericAgentParams map[string]any,
) (*SynthResult, error) {
if preferredProvider == "" {
preferredProvider = m.primary
}
providerNames := make([]string, 0, len(m.ttsProviders))
if _, ok := m.ttsProviders[preferredProvider]; ok {
providerNames = append(providerNames, preferredProvider)
}
if m.primary != preferredProvider {
if _, ok := m.ttsProviders[m.primary]; ok {
providerNames = append(providerNames, m.primary)
}
}
for name, p := range m.ttsProviders {
if name == m.primary {
continue
for name := range m.ttsProviders {
if name != preferredProvider && name != m.primary {
providerNames = append(providerNames, name)
}
}
var providerErrs []error
for _, name := range providerNames {
opts := TTSOptions{}
if resolveOpts != nil {
opts = resolveOpts(name)
}
attemptOpts := m.withAdaptedParams(opts, name, genericAgentParams)
p := m.ttsProviders[name]
result, err := p.Synthesize(ctx, text, attemptOpts)
if err == nil {
slog.Info("tts fallback succeeded", "provider", name)
return result, nil
}
slog.Warn("tts fallback provider failed", "provider", name, "error", err)
slog.Warn("tts provider failed, trying fallback", "provider", name, "error", err)
providerErrs = append(providerErrs, fmt.Errorf("%s: %w", name, err))
}
if len(providerErrs) == 0 {
+5 -10
View File
@@ -16,15 +16,10 @@ import (
"github.com/nextlevelbuilder/goclaw/internal/audio"
)
const (
// sttMaxBytes matches the OpenAI upload limit, which compatible engines
// generally adopt. Enforced before the request so an oversized file fails
// locally instead of after a full upload.
sttMaxBytes = 25 << 20 // 25 MB
// sttDefaultTimeout is the floor for transcription, which is slower than
// synthesis and routinely exceeds the shared 30 s default.
sttDefaultTimeout = 120 * time.Second
)
// sttMaxBytes matches the OpenAI upload limit, which compatible engines
// generally adopt. Enforced before the request so an oversized file fails
// locally instead of after a full upload.
const sttMaxBytes = 25 << 20 // 25 MB
// STTProvider transcribes audio via an OpenAI-compatible
// POST /audio/transcriptions.
@@ -67,7 +62,7 @@ func (p *STTProvider) Transcribe(ctx context.Context, in audio.STTInput, opts au
req.Header.Set("Authorization", "Bearer "+p.cfg.APIKey)
}
timeout := sttDefaultTimeout
timeout := time.Duration(p.cfg.TimeoutMs) * time.Millisecond
if opts.TimeoutMs > 0 {
timeout = time.Duration(opts.TimeoutMs) * time.Millisecond
}
+95
View File
@@ -8,6 +8,7 @@ import (
"os"
"path/filepath"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -202,6 +203,100 @@ func TestSTTOptionsModelOverridesConfig(t *testing.T) {
assert.Equal(t, "opts-model", got.model)
}
func TestSTTConfiguredTimeoutAppliesWithoutPerCallOverride(t *testing.T) {
requestStarted := make(chan struct{})
releaseServer := make(chan struct{})
srv := httptest.NewServer(http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) {
close(requestStarted)
select {
case <-releaseServer:
case <-r.Context().Done():
}
}))
t.Cleanup(srv.Close)
t.Cleanup(func() { close(releaseServer) })
provider, err := openaicompat.NewSTTProvider(openaicompat.Config{
APIBase: srv.URL,
TimeoutMs: 25,
})
require.NoError(t, err)
ctx, cancel := context.WithCancel(context.Background())
t.Cleanup(cancel)
result := make(chan error, 1)
go func() {
_, err := provider.Transcribe(ctx,
audio.STTInput{Bytes: []byte("x"), MimeType: "audio/wav"},
audio.STTOptions{})
result <- err
}()
select {
case <-requestStarted:
case <-time.After(time.Second):
t.Fatal("transcription request did not reach the test server")
}
select {
case err := <-result:
require.Error(t, err)
assert.ErrorIs(t, err, context.DeadlineExceeded)
case <-time.After(500 * time.Millisecond):
cancel()
<-result
t.Fatal("Transcribe did not honor Config.TimeoutMs")
}
}
func TestSTTOptionsTimeoutOverridesConfig(t *testing.T) {
requestStarted := make(chan struct{})
releaseResponse := make(chan struct{})
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
close(requestStarted)
select {
case <-releaseResponse:
_, _ = io.WriteString(w, `{"text":"ok"}`)
case <-r.Context().Done():
}
}))
t.Cleanup(srv.Close)
provider, err := openaicompat.NewSTTProvider(openaicompat.Config{
APIBase: srv.URL,
TimeoutMs: 25,
})
require.NoError(t, err)
result := make(chan error, 1)
go func() {
_, err := provider.Transcribe(context.Background(),
audio.STTInput{Bytes: []byte("x"), MimeType: "audio/wav"},
audio.STTOptions{TimeoutMs: 1000})
result <- err
}()
select {
case <-requestStarted:
case <-time.After(time.Second):
t.Fatal("transcription request did not reach the test server")
}
select {
case err := <-result:
t.Fatalf("Transcribe used Config.TimeoutMs instead of opts.TimeoutMs: %v", err)
case <-time.After(100 * time.Millisecond):
}
close(releaseResponse)
select {
case err := <-result:
require.NoError(t, err)
case <-time.After(time.Second):
t.Fatal("Transcribe did not return after the test server responded")
}
}
// TestSTTLanguageFallsBackToHint covers a plain (non-verbose) json response,
// which carries no language field.
func TestSTTLanguageFallsBackToHint(t *testing.T) {
+10 -3
View File
@@ -101,6 +101,10 @@ func (c *Channel) handleMessageEventFrom(ctx context.Context, event *MessageEven
// 3. Resolve sender name (cached)
senderName := c.resolveSenderName(ctx, mc.SenderID)
senderLabel := senderName
if senderLabel == "" {
senderLabel = mc.SenderID
}
// 4. Resolve media BEFORE mention gate so non-mentioned messages
// also have their files downloaded and stored in pending history.
@@ -166,7 +170,7 @@ func (c *Channel) handleMessageEventFrom(ctx context.Context, event *MessageEven
historyKey = fmt.Sprintf("%s:topic:%s", mc.ChatID, mc.RootID)
}
c.GroupHistory().Record(historyKey, channels.HistoryEntry{
Sender: senderName,
Sender: senderLabel,
SenderID: mc.SenderID,
Body: mc.Content,
Media: earlyMediaPaths,
@@ -307,9 +311,12 @@ func (c *Channel) handleMessageEventFrom(ctx context.Context, event *MessageEven
}
// Annotate content with sender identity so the agent knows who is messaging.
if senderName != "" {
if mc.ChatType == "group" || senderName != "" {
if mc.ChatType == "group" {
annotated := fmt.Sprintf("[From: %s]\n%s", senderName, content)
annotated := content
if senderLabel != "" {
annotated = fmt.Sprintf("[From: %s]\n%s", senderLabel, content)
}
if c.HistoryLimit() > 0 {
content = c.GroupHistory().BuildContext(chatID, annotated, c.HistoryLimit())
} else {
+135
View File
@@ -4,6 +4,7 @@ import (
"context"
"encoding/json"
"errors"
"strings"
"testing"
"time"
@@ -80,6 +81,140 @@ func TestCheckGroupPolicy_Pairing_GroupAllowListBypasses(t *testing.T) {
}
}
func TestHandleMessageEvent_GroupContextFallsBackToSenderIDWhenNameLookupFails(t *testing.T) {
requireMention := true
msgBus := bus.New()
defer msgBus.Close()
ch, err := New(config.FeishuConfig{
AppID: "app",
AppSecret: "secret",
GroupPolicy: "open",
RequireMention: &requireMention,
}, msgBus, nil, nil, nil)
if err != nil {
t.Fatalf("New: %v", err)
}
ch.SetName("feishu-test")
ch.botOpenID = "ou_target_bot"
srv := newSimpleMockServer(t, `{"code":50000,"msg":"name lookup failed"}`)
ch.client = NewLarkClient("app", "secret", srv.URL)
first := feishuGroupTextEvent(t, "om_context", "ou_duc", "oc_group", "Send AGENTS.md again", nil)
ch.handleMessageEvent(context.Background(), first)
assertNoFeishuInbound(t, msgBus)
entries := ch.GroupHistory().GetEntries("oc_group")
if len(entries) != 1 {
t.Fatalf("history entries = %d, want 1", len(entries))
}
if entries[0].Sender != "ou_duc" {
t.Fatalf("history sender = %q, want sender ID fallback", entries[0].Sender)
}
second := feishuGroupTextEvent(t, "om_mention", "ou_duc", "oc_group", "@_user_1", []EventMention{
feishuMention("@_user_1", "ou_target_bot", "Techlead"),
})
ch.handleMessageEvent(context.Background(), second)
msg := assertFeishuInbound(t, msgBus)
for _, want := range []string{
"[From: ou_duc]",
"[empty message]",
"Send AGENTS.md again",
} {
if !strings.Contains(msg.Content, want) {
t.Errorf("content missing %q:\n%s", want, msg.Content)
}
}
if got := ch.GroupHistory().GetEntries("oc_group"); len(got) != 0 {
t.Fatalf("history entries after publish = %d, want 0", len(got))
}
}
func TestHandleMessageEvent_GroupContextPreservesResolvedSenderName(t *testing.T) {
requireMention := true
msgBus := bus.New()
defer msgBus.Close()
ch, err := New(config.FeishuConfig{
AppID: "app",
AppSecret: "secret",
GroupPolicy: "open",
RequireMention: &requireMention,
}, msgBus, nil, nil, nil)
if err != nil {
t.Fatalf("New: %v", err)
}
ch.SetName("feishu-test")
ch.botOpenID = "ou_target_bot"
srv := newSimpleMockServer(t, `{"code":0,"msg":"ok","data":{"user":{"name":"Duc Nguyen"}}}`)
ch.client = NewLarkClient("app", "secret", srv.URL)
event := feishuGroupTextEvent(t, "om_named", "ou_duc", "oc_group", "@_user_1 hello", []EventMention{
feishuMention("@_user_1", "ou_target_bot", "Techlead"),
})
ch.handleMessageEvent(context.Background(), event)
msg := assertFeishuInbound(t, msgBus)
if msg.Content != "[From: Duc Nguyen]\nhello" {
t.Fatalf("content = %q, want resolved sender name", msg.Content)
}
if msg.Metadata["sender_name"] != "Duc Nguyen" {
t.Fatalf("sender_name metadata = %q", msg.Metadata["sender_name"])
}
}
func TestHandleMessageEvent_P2PContextRemainsNameBased(t *testing.T) {
msgBus := bus.New()
defer msgBus.Close()
ch, err := New(config.FeishuConfig{AppID: "app", AppSecret: "secret", DMPolicy: "open"}, msgBus, nil, nil, nil)
if err != nil {
t.Fatalf("New: %v", err)
}
ch.SetName("feishu-test")
srv := newSimpleMockServer(t, `{"code":0,"msg":"ok","data":{"user":{"name":"Duc Nguyen"}}}`)
ch.client = NewLarkClient("app", "secret", srv.URL)
event := feishuGroupTextEvent(t, "om_p2p", "ou_duc", "oc_p2p", "hello", nil)
event.Event.Message.ChatType = "p2p"
ch.handleMessageEvent(context.Background(), event)
msg := assertFeishuInbound(t, msgBus)
if msg.Content != "[From: Duc Nguyen]\nhello" {
t.Fatalf("content = %q, want unchanged p2p name context", msg.Content)
}
}
func TestHandleMessageEvent_GroupContextDoesNotEmitEmptySenderLabel(t *testing.T) {
requireMention := true
msgBus := bus.New()
defer msgBus.Close()
ch, err := New(config.FeishuConfig{
AppID: "app",
AppSecret: "secret",
GroupPolicy: "open",
RequireMention: &requireMention,
}, msgBus, nil, nil, nil)
if err != nil {
t.Fatalf("New: %v", err)
}
ch.SetName("feishu-test")
ch.botOpenID = "ou_target_bot"
event := feishuGroupTextEvent(t, "om_no_sender", "", "oc_group", "@_user_1", []EventMention{
feishuMention("@_user_1", "ou_target_bot", "Techlead"),
})
ch.handleMessageEvent(context.Background(), event)
msg := assertFeishuInbound(t, msgBus)
if msg.Content != "[empty message]" {
t.Fatalf("content = %q, want no empty sender annotation", msg.Content)
}
}
type feishuPolicyPairingStore struct {
requests int
lastSenderID string
+38 -73
View File
@@ -49,7 +49,41 @@ type feishuInstanceConfig struct {
// Factory creates a Feishu/Lark channel from DB instance data.
func Factory(name string, creds json.RawMessage, cfg json.RawMessage,
msgBus *bus.MessageBus, pairingSvc store.PairingStore) (channels.Channel, error) {
return buildChannel(name, creds, cfg, msgBus, pairingSvc, nil, nil, nil, nil)
}
// FactoryWithPendingStore returns a ChannelFactory with persistent history support.
func FactoryWithPendingStore(pendingStore store.PendingMessageStore) channels.ChannelFactory {
return FactoryWithPendingStoreAndAudio(pendingStore, nil)
}
// FactoryWithPendingStoreAndAudio returns a ChannelFactory with persistent history and STT support.
func FactoryWithPendingStoreAndAudio(pendingStore store.PendingMessageStore, audioMgr *audio.Manager) channels.ChannelFactory {
return FactoryWithStoresAndAudio(nil, nil, pendingStore, audioMgr)
}
// FactoryWithStores returns a ChannelFactory with the stores required by
// writer management and persistent group history.
func FactoryWithStores(agentStore store.AgentStore, configPermStore store.ConfigPermissionStore,
pendingStore store.PendingMessageStore) channels.ChannelFactory {
return FactoryWithStoresAndAudio(agentStore, configPermStore, pendingStore, nil)
}
// FactoryWithStoresAndAudio returns a ChannelFactory with writer management,
// persistent group history, and STT support.
func FactoryWithStoresAndAudio(agentStore store.AgentStore, configPermStore store.ConfigPermissionStore,
pendingStore store.PendingMessageStore, audioMgr *audio.Manager) channels.ChannelFactory {
return func(name string, creds json.RawMessage, cfg json.RawMessage,
msgBus *bus.MessageBus, pairingSvc store.PairingStore) (channels.Channel, error) {
return buildChannel(name, creds, cfg, msgBus, pairingSvc,
agentStore, configPermStore, pendingStore, audioMgr)
}
}
func buildChannel(name string, creds json.RawMessage, cfg json.RawMessage,
msgBus *bus.MessageBus, pairingSvc store.PairingStore,
agentStore store.AgentStore, configPermStore store.ConfigPermissionStore,
pendingStore store.PendingMessageStore, audioMgr *audio.Manager) (channels.Channel, error) {
var c feishuCreds
if len(creds) > 0 {
if err := json.Unmarshal(creds, &c); err != nil {
@@ -103,7 +137,10 @@ func Factory(name string, creds json.RawMessage, cfg json.RawMessage,
fsCfg.GroupPolicy = "pairing"
}
ch, err := New(fsCfg, msgBus, pairingSvc, nil, nil)
ch, err := New(fsCfg, msgBus, pairingSvc, pendingStore, audioMgr,
WithAgentStore(agentStore),
WithConfigPermStore(configPermStore),
)
if err != nil {
return nil, err
}
@@ -111,75 +148,3 @@ func Factory(name string, creds json.RawMessage, cfg json.RawMessage,
ch.SetName(name)
return ch, nil
}
// FactoryWithPendingStore returns a ChannelFactory with persistent history support.
func FactoryWithPendingStore(pendingStore store.PendingMessageStore) channels.ChannelFactory {
return FactoryWithPendingStoreAndAudio(pendingStore, nil)
}
// FactoryWithPendingStoreAndAudio returns a ChannelFactory with persistent history and STT support.
func FactoryWithPendingStoreAndAudio(pendingStore store.PendingMessageStore, audioMgr *audio.Manager) channels.ChannelFactory {
return func(name string, creds json.RawMessage, cfg json.RawMessage,
msgBus *bus.MessageBus, pairingSvc store.PairingStore) (channels.Channel, error) {
var c feishuCreds
if len(creds) > 0 {
if err := json.Unmarshal(creds, &c); err != nil {
return nil, fmt.Errorf("decode feishu credentials: %w", err)
}
}
if c.AppID == "" || c.AppSecret == "" {
return nil, fmt.Errorf("feishu app_id and app_secret are required")
}
var ic feishuInstanceConfig
if len(cfg) > 0 {
if err := json.Unmarshal(cfg, &ic); err != nil {
return nil, fmt.Errorf("decode feishu config: %w", err)
}
}
fsCfg := config.FeishuConfig{
Enabled: true,
AppID: c.AppID,
AppSecret: c.AppSecret,
EncryptKey: c.EncryptKey,
VerificationToken: c.VerificationToken,
Domain: ic.Domain,
ConnectionMode: ic.ConnectionMode,
WebhookPort: ic.WebhookPort,
WebhookPath: ic.WebhookPath,
AllowFrom: ic.AllowFrom,
DMPolicy: ic.DMPolicy,
GroupPolicy: ic.GroupPolicy,
GroupAllowFrom: ic.GroupAllowFrom,
RequireMention: ic.RequireMention,
TopicSessionMode: ic.TopicSessionMode,
TextChunkLimit: ic.TextChunkLimit,
MediaMaxMB: ic.MediaMaxMB,
RenderMode: ic.RenderMode,
Streaming: ic.Streaming,
ReactionLevel: ic.ReactionLevel,
HistoryLimit: ic.HistoryLimit,
BlockReply: ic.BlockReply,
ChatBehavior: ic.ChatBehavior,
STTProxyURL: ic.STTProxyURL,
STTAPIKey: ic.STTAPIKey,
STTTenantID: ic.STTTenantID,
STTTimeoutSeconds: ic.STTTimeoutSeconds,
VoiceAgentID: ic.VoiceAgentID,
}
if fsCfg.GroupPolicy == "" {
fsCfg.GroupPolicy = "pairing"
}
ch, err := New(fsCfg, msgBus, pairingSvc, pendingStore, audioMgr)
if err != nil {
return nil, err
}
ch.SetName(name)
return ch, nil
}
}
+80
View File
@@ -1,10 +1,28 @@
package feishu
import (
"context"
"encoding/json"
"testing"
"github.com/google/uuid"
"github.com/nextlevelbuilder/goclaw/internal/store"
)
type factoryAgentStore struct {
store.AgentStore
agent *store.AgentData
gotKey string
gotTenant uuid.UUID
}
func (f *factoryAgentStore) GetByKey(ctx context.Context, agentKey string) (*store.AgentData, error) {
f.gotKey = agentKey
f.gotTenant = store.TenantIDFromContext(ctx)
return f.agent, nil
}
func TestFactory_MissingAppID(t *testing.T) {
creds, _ := json.Marshal(map[string]string{"app_secret": "s"})
_, err := Factory("test", creds, nil, nil, nil)
@@ -141,3 +159,65 @@ func TestFactoryWithPendingStore_InvalidConfigJSON(t *testing.T) {
t.Fatal("expected error for invalid config JSON")
}
}
func TestFactoryWithStoresAndAudio_WiresWriterDependencies(t *testing.T) {
agentID := uuid.New()
tenantID := uuid.New()
agentStore := &factoryAgentStore{agent: &store.AgentData{
BaseModel: store.BaseModel{ID: agentID},
AgentKey: "writer-agent",
}}
configPermStore := &fakeConfigPermStore{}
factory := FactoryWithStoresAndAudio(agentStore, configPermStore, nil, nil)
creds, _ := json.Marshal(map[string]string{
"app_id": "cli_store_app",
"app_secret": "store_secret",
})
raw, err := factory("feishu-with-stores", creds, nil, nil, nil)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
ch, ok := raw.(*Channel)
if !ok {
t.Fatalf("factory returned %T, want *Channel", raw)
}
if ch.agentStore != agentStore {
t.Fatal("agent store was not wired into Feishu channel")
}
if ch.configPermStore != configPermStore {
t.Fatal("config permission store was not wired into Feishu channel")
}
srv, replies := captureReplies(t)
ch.client = NewLarkClient("app", "secret", srv.URL)
ch.botOpenID = "ou_fake_bot"
ch.SetAgentID("writer-agent")
ch.SetTenantID(tenantID)
handled := ch.maybeHandleWriterCommand(context.Background(), &messageContext{
ChatID: "oc_factory_group",
MessageID: "om_factory_command",
SenderID: "ou_alice",
ChatType: "group",
Content: "/addwriter @_user_1",
Mentions: []mentionInfo{{Key: "@_user_1", OpenID: "ou_alice", Name: "Alice"}},
})
if !handled {
t.Fatal("expected /addwriter to be handled")
}
assertReplyContains(t, *replies, "Added Alice as a file writer")
if agentStore.gotKey != "writer-agent" {
t.Fatalf("agent lookup key = %q, want writer-agent", agentStore.gotKey)
}
if agentStore.gotTenant != tenantID {
t.Fatalf("agent lookup tenant = %s, want %s", agentStore.gotTenant, tenantID)
}
writers, err := configPermStore.ListFileWriters(context.Background(), agentID, "group:feishu-with-stores:oc_factory_group")
if err != nil {
t.Fatalf("list file writers: %v", err)
}
if len(writers) != 1 || writers[0].UserID != "ou_alice" {
t.Fatalf("writers = %+v, want ou_alice", writers)
}
}
+6
View File
@@ -7,6 +7,7 @@ import (
"fmt"
"log/slog"
"maps"
"strings"
"sync"
"sync/atomic"
"time"
@@ -198,6 +199,7 @@ func (d *stdDispatcher) runSync(ctx context.Context, ev Event, chain []HookConfi
evMut.ToolInput = cloneMap(ev.ToolInput)
}
mutated := false
additionalContexts := make([]string, 0, len(chain))
for _, cfg := range chain {
if !cfg.Enabled {
@@ -259,6 +261,9 @@ func (d *stdDispatcher) runSync(ctx context.Context, ev Event, chain []HookConfi
)
}
}
if dec == DecisionAllow && scriptRes.AdditionalContext != "" {
additionalContexts = append(additionalContexts, scriptRes.AdditionalContext)
}
switch dec {
case DecisionBlock:
@@ -282,6 +287,7 @@ func (d *stdDispatcher) runSync(ctx context.Context, ev Event, chain []HookConfi
}
result := FireResult{Decision: DecisionAllow}
result.AdditionalContext = strings.Join(additionalContexts, "\n\n")
if mutated {
if evMut.ToolInput != nil {
result.UpdatedToolInput = evMut.ToolInput
+30 -1
View File
@@ -411,16 +411,45 @@ func TestNoopDispatcher_AlwaysAllows(t *testing.T) {
// the dispatcher's source-tier gate can be exercised without invoking the
// real goja runtime.
type mutatingHandler struct {
updated map[string]any
updated map[string]any
additionalContext string
}
func (h *mutatingHandler) Execute(ctx context.Context, _ hooks.HookConfig, _ hooks.Event) (hooks.Decision, error) {
if r := hooks.ScriptResultFrom(ctx); r != nil {
r.UpdatedInput = h.updated
r.AdditionalContext = h.additionalContext
}
return hooks.DecisionAllow, nil
}
func TestDispatcher_UIScriptAdditionalContextAllowedWithoutInputMutation(t *testing.T) {
cfg := newBaseHook(hooks.HandlerScript, hooks.EventUserPromptSubmit)
cfg.Source = "ui"
fs := &fakeStore{hooks: []hooks.HookConfig{cfg}}
h := &mutatingHandler{
updated: map[string]any{"rawInput": "must-not-apply"},
additionalContext: "Call the required workflow tool.",
}
d := hooks.NewStdDispatcher(hooks.StdDispatcherOpts{
Store: fs, Audit: hooks.NewAuditWriter(fs, ""),
Handlers: map[hooks.HandlerType]hooks.Handler{hooks.HandlerScript: h},
})
r, err := d.Fire(context.Background(), hooks.Event{
EventID: "e-context", HookEvent: hooks.EventUserPromptSubmit, RawInput: "original",
})
if err != nil || r.Decision != hooks.DecisionAllow {
t.Fatalf("decision=%q err=%v", r.Decision, err)
}
if r.AdditionalContext != "Call the required workflow tool." {
t.Fatalf("additional context=%q", r.AdditionalContext)
}
if r.UpdatedRawInput != nil {
t.Fatalf("UI source mutated raw input: %q", *r.UpdatedRawInput)
}
}
// TestDispatcher_ScriptMutation_BuiltinSourceApplies verifies the source-tier
// capability: a script hook with source="builtin" has its UpdatedInput applied
// to FireResult.Updated*; a script hook with source="ui" does NOT.
+18 -7
View File
@@ -24,11 +24,11 @@ import (
// ── Public surface ──────────────────────────────────────────────────────────
// ProviderResolver returns a provider + resolved model name for a given
// (tenantID, preferredModel). preferredModel is the UI/config-specified
// model (e.g. "haiku"); resolver may expand aliases or fall back to the
// tenant's default when the alias is unknown.
// tenant. preferredProvider and preferredModel come from hook config. Legacy
// hooks may omit preferredProvider, in which case the resolver applies its
// backward-compatible alias/config fallback chain.
type ProviderResolver interface {
ResolveForHook(ctx context.Context, tenantID uuid.UUID, preferredModel string) (providers.Provider, string, error)
ResolveForHook(ctx context.Context, tenantID uuid.UUID, preferredProvider, preferredModel string) (providers.Provider, string, error)
}
// PromptHandler implements hooks.Handler via an LLM structured-output call.
@@ -150,8 +150,9 @@ func (h *PromptHandler) Execute(ctx context.Context, cfg hooks.HookConfig, ev ho
}
// 4. Resolve provider
model := h.modelFor(cfg)
provider, resolvedModel, err := h.Resolver.ResolveForHook(ctx, ev.TenantID, model)
providerName := h.providerFor(cfg)
model := h.modelFor(cfg, providerName != "")
provider, resolvedModel, err := h.Resolver.ResolveForHook(ctx, ev.TenantID, providerName, model)
if err != nil || provider == nil {
return hooks.DecisionError, fmt.Errorf("hook: prompt handler: resolve provider: %w", err)
}
@@ -233,10 +234,20 @@ func (h *PromptHandler) maxInvocations(cfg hooks.HookConfig) int {
return defaultPromptMaxInvocations
}
func (h *PromptHandler) modelFor(cfg hooks.HookConfig) string {
func (h *PromptHandler) providerFor(cfg hooks.HookConfig) string {
provider, _ := cfg.Config["provider"].(string)
return strings.TrimSpace(provider)
}
func (h *PromptHandler) modelFor(cfg hooks.HookConfig, hasExplicitProvider bool) string {
if m, _ := cfg.Config["model"].(string); m != "" {
return m
}
// An explicit provider with no model means "use that provider's default".
// Only legacy model-only hooks inherit the historical Haiku fallback.
if hasExplicitProvider {
return ""
}
if h.DefaultModel != "" {
return h.DefaultModel
}
+33 -13
View File
@@ -3,6 +3,7 @@ package handlers
import (
"context"
"errors"
"fmt"
"strings"
"github.com/google/uuid"
@@ -15,17 +16,18 @@ import (
// the ProviderResolver interface consumed by PromptHandler. It applies a
// simple fallback chain:
//
// 1. Explicit model alias → map to provider by convention (haiku/sonnet/opus → anthropic).
// 2. System config `hooks.prompt.provider` / `hooks.prompt.model`.
// 3. System config `background.provider` / `background.model`.
// 4. First registered provider for the tenant.
// 1. Explicit provider + model from the hook config.
// 2. Legacy explicit model alias → map to provider by convention.
// 3. System config `hooks.prompt.provider` / `hooks.prompt.model`.
// 4. System config `background.provider` / `background.model`.
// 5. First registered provider for the tenant.
//
// Keeping this adapter in `handlers` rather than the higher-level
// `providerresolve` package avoids a new import cycle (providerresolve →
// store → hooks would introduce a diamond).
type RegistryResolver struct {
Registry *providers.Registry
SysConfig store.SystemConfigStore
Registry *providers.Registry
SysConfig store.SystemConfigStore
DefaultProviderForAlias func(alias string) string
}
@@ -33,19 +35,36 @@ type RegistryResolver struct {
// registry MUST be non-nil. sysConfig may be nil (fallback to step 4 only).
func NewRegistryResolver(registry *providers.Registry, sysConfig store.SystemConfigStore) *RegistryResolver {
return &RegistryResolver{
Registry: registry,
SysConfig: sysConfig,
Registry: registry,
SysConfig: sysConfig,
DefaultProviderForAlias: defaultProviderForAlias,
}
}
// ResolveForHook implements ProviderResolver.
func (r *RegistryResolver) ResolveForHook(ctx context.Context, tenantID uuid.UUID, preferredModel string) (providers.Provider, string, error) {
func (r *RegistryResolver) ResolveForHook(ctx context.Context, tenantID uuid.UUID, preferredProvider, preferredModel string) (providers.Provider, string, error) {
if r == nil || r.Registry == nil {
return nil, "", errors.New("hook resolver: nil registry")
}
// Step 1: explicit alias → provider name.
// Step 1: an explicit provider is authoritative. Never silently route its
// model to another provider when a hook is misconfigured.
if providerName := strings.TrimSpace(preferredProvider); providerName != "" {
p, err := r.Registry.GetForTenant(tenantID, providerName)
if err != nil {
return nil, "", fmt.Errorf("hook resolver: configured provider %q: %w", providerName, err)
}
if p == nil {
return nil, "", fmt.Errorf("hook resolver: configured provider %q returned nil", providerName)
}
model := strings.TrimSpace(preferredModel)
if model == "" {
model = p.DefaultModel()
}
return p, model, nil
}
// Step 2: legacy explicit alias → provider name.
if preferredModel != "" {
if name := r.providerForAlias(preferredModel); name != "" {
if p, err := r.Registry.GetForTenant(tenantID, name); err == nil && p != nil {
@@ -54,18 +73,19 @@ func (r *RegistryResolver) ResolveForHook(ctx context.Context, tenantID uuid.UUI
}
}
// Step 2: system config hooks.prompt.*
// Step 3: system config hooks.prompt.*
configs := r.loadConfigs(ctx, tenantID)
if p, m, ok := r.tryConfig(tenantID, configs["hooks.prompt.provider"], configs["hooks.prompt.model"], preferredModel); ok {
return p, m, nil
}
// Step 3: fall back to background.*
// Step 4: fall back to background.*
if p, m, ok := r.tryConfig(tenantID, configs["background.provider"], configs["background.model"], preferredModel); ok {
return p, m, nil
}
// Step 4: first registered provider for the tenant.
// Step 5: first registered provider for the tenant. This path remains only
// for backward compatibility with hooks that predate explicit providers.
names := r.Registry.ListForTenant(tenantID)
if len(names) == 0 {
return nil, "", errors.New("hook resolver: no providers registered for tenant")
@@ -0,0 +1,72 @@
package handlers_test
import (
"context"
"testing"
"github.com/google/uuid"
"github.com/nextlevelbuilder/goclaw/internal/hooks/handlers"
"github.com/nextlevelbuilder/goclaw/internal/providers"
)
func TestRegistryResolver_ExplicitProviderIsAuthoritative(t *testing.T) {
tenantID := uuid.New()
registry := providers.NewRegistry(nil)
registry.RegisterForTenant(tenantID, &fakeProvider{name: "cppai", defaultModel: "gpt-default"})
registry.RegisterForTenant(tenantID, &fakeProvider{name: "anthropic", defaultModel: "claude-default"})
resolver := handlers.NewRegistryResolver(registry, nil)
provider, model, err := resolver.ResolveForHook(context.Background(), tenantID, "cppai", "gpt-5.6-terra")
if err != nil {
t.Fatalf("ResolveForHook() error: %v", err)
}
if provider.Name() != "cppai" || model != "gpt-5.6-terra" {
t.Fatalf("resolved=(%q, %q), want (cppai, gpt-5.6-terra)", provider.Name(), model)
}
}
func TestRegistryResolver_ExplicitProviderUsesItsDefaultModel(t *testing.T) {
tenantID := uuid.New()
registry := providers.NewRegistry(nil)
registry.RegisterForTenant(tenantID, &fakeProvider{name: "bailian", defaultModel: "qwen3.7-plus"})
resolver := handlers.NewRegistryResolver(registry, nil)
provider, model, err := resolver.ResolveForHook(context.Background(), tenantID, "bailian", "")
if err != nil {
t.Fatalf("ResolveForHook() error: %v", err)
}
if provider.Name() != "bailian" || model != "qwen3.7-plus" {
t.Fatalf("resolved=(%q, %q), want (bailian, qwen3.7-plus)", provider.Name(), model)
}
}
func TestRegistryResolver_MissingExplicitProviderDoesNotFallback(t *testing.T) {
tenantID := uuid.New()
registry := providers.NewRegistry(nil)
registry.RegisterForTenant(tenantID, &fakeProvider{name: "bailian", defaultModel: "qwen3.7-plus"})
resolver := handlers.NewRegistryResolver(registry, nil)
provider, model, err := resolver.ResolveForHook(context.Background(), tenantID, "missing", "gpt-5.6-terra")
if err == nil {
t.Fatal("ResolveForHook() error=nil, want configured-provider error")
}
if provider != nil || model != "" {
t.Fatalf("resolved=(%v, %q), want nil provider and empty model", provider, model)
}
}
func TestRegistryResolver_LegacyModelAliasStillResolves(t *testing.T) {
tenantID := uuid.New()
registry := providers.NewRegistry(nil)
registry.RegisterForTenant(tenantID, &fakeProvider{name: "anthropic", defaultModel: "claude-default"})
resolver := handlers.NewRegistryResolver(registry, nil)
provider, model, err := resolver.ResolveForHook(context.Background(), tenantID, "", "haiku")
if err != nil {
t.Fatalf("ResolveForHook() error: %v", err)
}
if provider.Name() != "anthropic" || model != "haiku" {
t.Fatalf("resolved=(%q, %q), want (anthropic, haiku)", provider.Name(), model)
}
}
+45 -5
View File
@@ -17,14 +17,18 @@ import (
// fakeResolver returns a static provider + model. Counts resolve calls for
// cache-hit assertions.
type fakeResolver struct {
prov providers.Provider
model string
calls atomic.Int32
resolveErr error
prov providers.Provider
model string
calls atomic.Int32
resolveErr error
preferredProvider string
preferredModel string
}
func (f *fakeResolver) ResolveForHook(_ context.Context, _ uuid.UUID, _ string) (providers.Provider, string, error) {
func (f *fakeResolver) ResolveForHook(_ context.Context, _ uuid.UUID, preferredProvider, preferredModel string) (providers.Provider, string, error) {
f.calls.Add(1)
f.preferredProvider = preferredProvider
f.preferredModel = preferredModel
if f.resolveErr != nil {
return nil, "", f.resolveErr
}
@@ -115,6 +119,42 @@ func TestPrompt_Allow_StructuredOutput(t *testing.T) {
}
}
func TestPrompt_ExplicitProviderPassedToResolver(t *testing.T) {
prov := &fakeProvider{nextResp: okResp("allow")}
resolver := &fakeResolver{prov: prov, model: "gpt-5.6-terra"}
h := &handlers.PromptHandler{Resolver: resolver}
cfg := makePromptCfg(t)
cfg.Config["provider"] = "cppai"
cfg.Config["model"] = "gpt-5.6-terra"
dec, err := h.Execute(context.Background(), cfg, makePromptEv())
if err != nil {
t.Fatalf("unexpected err: %v", err)
}
if dec != hooks.DecisionAllow {
t.Fatalf("decision=%q, want allow", dec)
}
if resolver.preferredProvider != "cppai" || resolver.preferredModel != "gpt-5.6-terra" {
t.Fatalf("resolver preferences=(%q, %q), want (cppai, gpt-5.6-terra)", resolver.preferredProvider, resolver.preferredModel)
}
}
func TestPrompt_ExplicitProviderWithoutModelUsesProviderDefault(t *testing.T) {
prov := &fakeProvider{nextResp: okResp("allow")}
resolver := &fakeResolver{prov: prov, model: "provider-default"}
h := &handlers.PromptHandler{Resolver: resolver, DefaultModel: "haiku"}
cfg := makePromptCfg(t)
cfg.Config["provider"] = "cppai"
delete(cfg.Config, "model")
if _, err := h.Execute(context.Background(), cfg, makePromptEv()); err != nil {
t.Fatalf("unexpected err: %v", err)
}
if resolver.preferredModel != "" {
t.Fatalf("preferred model=%q, want empty provider-default sentinel", resolver.preferredModel)
}
}
func TestPrompt_Block_StructuredOutput(t *testing.T) {
prov := &fakeProvider{nextResp: okResp("block")}
h := &handlers.PromptHandler{Resolver: &fakeResolver{prov: prov, model: "m"}}
+2 -1
View File
@@ -187,7 +187,7 @@ func (h *ScriptHandler) Execute(ctx context.Context, cfg hooks.HookConfig, ev ho
return hooks.DecisionError, sanitizeError(execErr)
}
dec, reason, updated, parseErr := parseReturn(rt, result)
dec, reason, additionalContext, updated, parseErr := parseReturn(rt, result)
if parseErr != nil {
return hooks.DecisionError, parseErr
}
@@ -208,6 +208,7 @@ func (h *ScriptHandler) Execute(ctx context.Context, cfg hooks.HookConfig, ev ho
}
if r := hooks.ScriptResultFrom(ctx); r != nil {
r.Reason = reason
r.AdditionalContext = additionalContext
r.UpdatedInput = updated
r.Stdout = stdout.String()
}
+15 -8
View File
@@ -134,31 +134,32 @@ func captureStdout(rt *goja.Runtime, buf *strings.Builder) {
_ = rt.Set("console", console)
}
// parseReturn enforces the `{decision, reason?, updatedInput?}` contract.
// parseReturn enforces the
// `{decision, reason?, additionalContext?, updatedInput?}` contract.
// Any deviation (non-object, missing field, unknown decision string) maps to
// DecisionError with a descriptive error.
func parseReturn(rt *goja.Runtime, v goja.Value) (hooks.Decision, string, map[string]any, error) {
func parseReturn(rt *goja.Runtime, v goja.Value) (hooks.Decision, string, string, map[string]any, error) {
if v == nil || goja.IsUndefined(v) || goja.IsNull(v) {
return hooks.DecisionError, "", nil, errors.New("script return: must be object with decision")
return hooks.DecisionError, "", "", nil, errors.New("script return: must be object with decision")
}
obj := v.ToObject(rt)
if obj == nil {
return hooks.DecisionError, "", nil, errors.New("script return: must be object")
return hooks.DecisionError, "", "", nil, errors.New("script return: must be object")
}
decRaw := obj.Get("decision")
if decRaw == nil || goja.IsUndefined(decRaw) {
return hooks.DecisionError, "", nil, errors.New("script return: missing decision field")
return hooks.DecisionError, "", "", nil, errors.New("script return: missing decision field")
}
decStr, ok := decRaw.Export().(string)
if !ok {
return hooks.DecisionError, "", nil, errors.New("script return: decision must be string")
return hooks.DecisionError, "", "", nil, errors.New("script return: decision must be string")
}
dec := hooks.Decision(decStr)
switch dec {
case hooks.DecisionAllow, hooks.DecisionBlock, hooks.DecisionAsk, hooks.DecisionDefer:
// ok
default:
return hooks.DecisionError, "", nil, fmt.Errorf("script return: invalid decision %q", decStr)
return hooks.DecisionError, "", "", nil, fmt.Errorf("script return: invalid decision %q", decStr)
}
var reason string
if r := obj.Get("reason"); r != nil && !goja.IsUndefined(r) {
@@ -166,11 +167,17 @@ func parseReturn(rt *goja.Runtime, v goja.Value) (hooks.Decision, string, map[st
reason = s
}
}
var additionalContext string
if a := obj.Get("additionalContext"); a != nil && !goja.IsUndefined(a) {
if s, ok := a.Export().(string); ok {
additionalContext = s
}
}
var updated map[string]any
if u := obj.Get("updatedInput"); u != nil && !goja.IsUndefined(u) {
if m, ok := u.Export().(map[string]any); ok {
updated = m
}
}
return dec, reason, updated, nil
return dec, reason, additionalContext, updated, nil
}
+12
View File
@@ -73,6 +73,18 @@ func TestHappyPathAllow(t *testing.T) {
}
}
func TestAllowReturnsAdditionalContext(t *testing.T) {
src := `function handle(event) { return {decision: "allow", additionalContext: "Use the required workflow."}; }`
h := newTestHandler()
dec, err, res := runWithResult(t, h, mkCfg(src), mkEvent(), 500*time.Millisecond)
if err != nil || dec != hooks.DecisionAllow {
t.Fatalf("decision=%v err=%v", dec, err)
}
if res.AdditionalContext != "Use the required workflow." {
t.Fatalf("additional context=%q", res.AdditionalContext)
}
}
// TestHappyPathBlock verifies block + reason round-trips.
func TestHappyPathBlock(t *testing.T) {
src := `function handle(event) { return {decision: "block", reason: "nope"}; }`
+5 -1
View File
@@ -3,7 +3,7 @@ package hooks
import "context"
// ScriptResult carries non-decision outputs from a script handler execution
// (reason text, updatedInput proposal, captured stdout). The dispatcher
// (reason text, additional context, updatedInput proposal, captured stdout). The dispatcher
// provisions one via context before calling Handler.Execute; the handler
// populates it on success. Phase 03 wires this into the dispatcher so that
// only builtin-source hooks can apply UpdatedInput to the pipeline state.
@@ -13,6 +13,10 @@ import "context"
type ScriptResult struct {
// Reason is the human-readable explanation returned by the script.
Reason string
// AdditionalContext is instruction text appended to the current run's
// extra system prompt after the hook returns allow. It does not mutate the
// original user input.
AdditionalContext string
// UpdatedInput is a proposed replacement for Event.ToolInput. Dispatcher
// applies only when cfg.Source == "builtin" (source-tier capability).
UpdatedInput map[string]any
+7 -7
View File
@@ -151,12 +151,12 @@ func (d Decision) IsBlock() bool {
// DecisionReason carries a human-readable explanation when Decision is
// DecisionBlock (or DecisionAsk/DecisionDefer treated as block). This is
// injected as a user message to trigger a retry iteration.
//
type FireResult struct {
Decision Decision
DecisionReason string
UpdatedToolInput map[string]any
UpdatedRawInput *string
Decision Decision
DecisionReason string
AdditionalContext string
UpdatedToolInput map[string]any
UpdatedRawInput *string
}
// ─── Config & execution structs ──────────────────────────────────────────────
@@ -233,7 +233,7 @@ type Event struct {
HookEvent HookEvent
// PostModelResponse fields (populated when HookEvent == EventPostModelResponse).
ModelResponse string // the generated response content
Thinking string // reasoning content (if any)
ModelResponse string // the generated response content
Thinking string // reasoning content (if any)
ToolCalls []providers.ToolCall // empty for final response, populated if tool calls present
}
+39 -7
View File
@@ -10,6 +10,7 @@ import (
"math"
"net/http"
"strings"
"time"
)
// ContentHash returns a short SHA256 hex digest of the content (first 16 bytes).
@@ -144,6 +145,7 @@ type OpenAIEmbeddingProvider struct {
apiKey string
apiURL string
dimensions int // optional: truncate output to this many dimensions (0 = use model default)
httpClient *http.Client
}
// NewOpenAIEmbeddingProvider creates a provider for OpenAI-compatible embedding APIs.
@@ -156,10 +158,11 @@ func NewOpenAIEmbeddingProvider(name, apiKey, apiURL, model string) *OpenAIEmbed
}
return &OpenAIEmbeddingProvider{
name: name,
model: model,
apiKey: apiKey,
apiURL: apiURL,
name: name,
model: model,
apiKey: apiKey,
apiURL: apiURL,
httpClient: &http.Client{Timeout: 60 * time.Second},
}
}
@@ -194,7 +197,7 @@ func (p *OpenAIEmbeddingProvider) Embed(ctx context.Context, texts []string) ([]
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", "Bearer "+p.apiKey)
resp, err := http.DefaultClient.Do(req)
resp, err := p.httpClient.Do(req)
if err != nil {
return nil, fmt.Errorf("embedding request: %w", err)
}
@@ -208,6 +211,7 @@ func (p *OpenAIEmbeddingProvider) Embed(ctx context.Context, texts []string) ([]
var result struct {
Data []struct {
Embedding []float32 `json:"embedding"`
Index *int `json:"index"`
} `json:"data"`
}
@@ -215,9 +219,37 @@ func (p *OpenAIEmbeddingProvider) Embed(ctx context.Context, texts []string) ([]
return nil, fmt.Errorf("decode response: %w", err)
}
if len(result.Data) != len(texts) {
return nil, fmt.Errorf("embedding response count: got %d vectors for %d inputs", len(result.Data), len(texts))
}
embeddings := make([][]float32, len(result.Data))
for i, d := range result.Data {
embeddings[i] = d.Embedding
indexed := false
for _, d := range result.Data {
indexed = indexed || d.Index != nil
}
seen := make([]bool, len(result.Data))
for responseIndex, d := range result.Data {
targetIndex := responseIndex
if indexed {
if d.Index == nil || *d.Index < 0 || *d.Index >= len(result.Data) || seen[*d.Index] {
return nil, fmt.Errorf("embedding response has invalid index at position %d", responseIndex)
}
targetIndex = *d.Index
}
seen[targetIndex] = true
if len(d.Embedding) == 0 {
return nil, fmt.Errorf("embedding response has empty vector at index %d", targetIndex)
}
if p.dimensions > 0 && len(d.Embedding) != p.dimensions {
return nil, fmt.Errorf("embedding response dimension at index %d: got %d, want %d", targetIndex, len(d.Embedding), p.dimensions)
}
for _, value := range d.Embedding {
if math.IsNaN(float64(value)) || math.IsInf(float64(value), 0) {
return nil, fmt.Errorf("embedding response has non-finite value at index %d", targetIndex)
}
}
embeddings[targetIndex] = d.Embedding
}
return embeddings, nil
@@ -0,0 +1,52 @@
package memory
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
"time"
)
func TestOpenAIEmbeddingProviderRestoresResponseOrder(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(map[string]any{
"data": []map[string]any{
{"index": 1, "embedding": []float32{2, 2, 2}},
{"index": 0, "embedding": []float32{1, 1, 1}},
},
})
}))
defer server.Close()
provider := NewOpenAIEmbeddingProvider("test", "key", server.URL, "model").WithDimensions(3)
embeddings, err := provider.Embed(context.Background(), []string{"first", "second"})
if err != nil {
t.Fatalf("Embed() error = %v", err)
}
if embeddings[0][0] != 1 || embeddings[1][0] != 2 {
t.Fatalf("embeddings returned in response order: %v", embeddings)
}
}
func TestOpenAIEmbeddingProviderRejectsWrongDimension(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"data":[{"index":0,"embedding":[1,2]}]}`))
}))
defer server.Close()
provider := NewOpenAIEmbeddingProvider("test", "key", server.URL, "model").WithDimensions(3)
if _, err := provider.Embed(context.Background(), []string{"text"}); err == nil {
t.Fatal("Embed() error = nil, want dimension validation error")
}
}
func TestOpenAIEmbeddingProviderUsesBoundedHTTPClient(t *testing.T) {
provider := NewOpenAIEmbeddingProvider("test", "key", "http://example.invalid", "model")
if provider.httpClient == nil || provider.httpClient.Timeout != 60*time.Second {
t.Fatalf("HTTP timeout = %v, want 60s", provider.httpClient)
}
}
+10 -2
View File
@@ -60,8 +60,16 @@ func (s *ContextStage) Execute(ctx context.Context, state *RunState) error {
HookEvent: hooks.EventUserPromptSubmit,
}); r.Decision == hooks.DecisionBlock {
return fmt.Errorf("hook blocked user_prompt_submit")
} else if r.UpdatedRawInput != nil {
state.Input.Message = *r.UpdatedRawInput
} else {
if r.UpdatedRawInput != nil {
state.Input.Message = *r.UpdatedRawInput
}
if r.AdditionalContext != "" {
if state.Input.ExtraSystemPrompt != "" {
state.Input.ExtraSystemPrompt += "\n\n"
}
state.Input.ExtraSystemPrompt += r.AdditionalContext
}
}
// 0. Inject context values (agent/tenant/user/workspace scoping, input guard, truncation).
+26
View File
@@ -2631,6 +2631,32 @@ func TestFinalizeStage_PopulatesFileSizes_SkipsNonexistent(t *testing.T) {
// --- ContextStage tests ---
func TestContextStage_AppendsHookAdditionalContextToSystemPromptInput(t *testing.T) {
t.Parallel()
var capturedExtra string
deps := &PipelineDeps{
Hooks: toolStageHookDispatcher{fire: func(_ context.Context, ev hooks.Event) (hooks.FireResult, error) {
if ev.HookEvent == hooks.EventUserPromptSubmit {
return hooks.FireResult{Decision: hooks.DecisionAllow, AdditionalContext: "mandatory workflow"}, nil
}
return hooks.FireResult{Decision: hooks.DecisionAllow}, nil
}},
BuildMessages: func(_ context.Context, input *RunInput, _ []providers.Message, _ string) ([]providers.Message, error) {
capturedExtra = input.ExtraSystemPrompt
return []providers.Message{{Role: "system", Content: input.ExtraSystemPrompt}}, nil
},
}
state := defaultState()
state.Input.ExtraSystemPrompt = "existing context"
stage := NewContextStage(deps)
if err := stage.Execute(context.Background(), state); err != nil {
t.Fatalf("Execute() error: %v", err)
}
if capturedExtra != "existing context\n\nmandatory workflow" {
t.Fatalf("extra system prompt=%q", capturedExtra)
}
}
func TestContextStage_ResolveWorkspace(t *testing.T) {
t.Parallel()
wantWS := &workspace.WorkspaceContext{ActivePath: "/resolved"}
+92 -40
View File
@@ -4,6 +4,7 @@ import (
"context"
"database/sql"
"encoding/json"
"errors"
"fmt"
"log/slog"
"strings"
@@ -18,10 +19,11 @@ import (
type PGAgentStore struct {
db *sql.DB
embProvider store.EmbeddingProvider // optional: for agent frontmatter embeddings
embJobs chan struct{}
}
func NewPGAgentStore(db *sql.DB) *PGAgentStore {
return &PGAgentStore{db: db}
return &PGAgentStore{db: db, embJobs: make(chan struct{}, 4)}
}
// SetEmbeddingProvider sets the embedding provider for agent frontmatter vectors.
@@ -44,46 +46,96 @@ func (s *PGAgentStore) generateAgentEmbedding(ctx context.Context, agentID uuid.
return
}
vecStr := vectorToString(embeddings[0])
if _, err := s.db.ExecContext(ctx, `UPDATE agents SET embedding = $1::vector WHERE id = $2`, vecStr, agentID); err != nil {
if _, err := s.db.ExecContext(ctx, `
UPDATE agents SET embedding = $1::vector
WHERE id = $2 AND deleted_at IS NULL AND status = 'active'
AND COALESCE(display_name, '') = $3
AND COALESCE(frontmatter, '') = $4`,
vecStr, agentID, displayName, frontmatter); err != nil {
slog.Warn("agent embedding update failed", "agent", agentID, "error", err)
}
}
// queueAgentEmbedding bounds background provider calls. Updates clear the old
// vector before enqueueing, so overload leaves a retryable NULL instead of a
// stale semantic index.
func (s *PGAgentStore) queueAgentEmbedding(agentID uuid.UUID) {
if s.embProvider == nil {
return
}
select {
case s.embJobs <- struct{}{}:
go func() {
defer func() { <-s.embJobs }()
ctx, cancel := context.WithTimeout(store.WithCrossTenant(context.Background()), 75*time.Second)
defer cancel()
agent, err := s.GetByID(ctx, agentID)
if err != nil {
slog.Warn("agent embedding source load failed", "agent", agentID, "error", err)
return
}
s.generateAgentEmbedding(ctx, agent.ID, agent.DisplayName, agent.Frontmatter)
}()
default:
slog.Warn("agent embedding queue full; vector remains pending", "agent", agentID)
}
}
// BackfillAgentEmbeddings generates embeddings for all active agents that have frontmatter but no embedding.
func (s *PGAgentStore) BackfillAgentEmbeddings(ctx context.Context) (int, error) {
if s.embProvider == nil {
return 0, nil
}
var pending []agentBackfillRow
if err := pkgSqlxDB.SelectContext(ctx, &pending,
`SELECT id, COALESCE(display_name, '') AS display_name, COALESCE(frontmatter, '') AS frontmatter
FROM agents WHERE deleted_at IS NULL AND frontmatter IS NOT NULL AND frontmatter != '' AND embedding IS NULL`,
); err != nil {
return 0, err
}
if len(pending) == 0 {
return 0, nil
return 0, fmt.Errorf("no embedding provider configured")
}
slog.Info("backfilling agent embeddings", "count", len(pending))
updated := 0
for _, ag := range pending {
text := ag.DisplayName
if ag.Frontmatter != "" {
text += ": " + ag.Frontmatter
const batchSize = 50
total := 0
var backfillErrs []error
cursor := uuid.Nil
for {
var pending []agentBackfillRow
if err := pkgSqlxDB.SelectContext(ctx, &pending,
`SELECT id, COALESCE(display_name, '') AS display_name, COALESCE(frontmatter, '') AS frontmatter
FROM agents
WHERE deleted_at IS NULL AND status = 'active'
AND frontmatter IS NOT NULL AND frontmatter != '' AND embedding IS NULL AND id > $1
ORDER BY id ASC
LIMIT $2`, cursor, batchSize,
); err != nil {
return total, fmt.Errorf("query agents without embeddings: %w", err)
}
embeddings, err := s.embProvider.Embed(ctx, []string{text})
if err != nil || len(embeddings) == 0 || len(embeddings[0]) == 0 {
continue
if len(pending) == 0 {
return total, errors.Join(backfillErrs...)
}
vecStr := vectorToString(embeddings[0])
if _, err := s.db.ExecContext(ctx, `UPDATE agents SET embedding = $1::vector WHERE id = $2`, vecStr, ag.ID); err != nil {
continue
items := make([]embeddingBackfillItem[agentBackfillRow], len(pending))
for i, agent := range pending {
items[i] = embeddingBackfillItem[agentBackfillRow]{
row: agent,
id: agent.ID,
text: agent.DisplayName + ": " + agent.Frontmatter,
}
}
updated++
updated, err := processEmbeddingBackfillBatch(ctx, s.embProvider, "agent", items,
func(ctx context.Context, agent agentBackfillRow, embedding []float32) (int64, error) {
result, err := s.db.ExecContext(ctx,
`UPDATE agents SET embedding = $1::vector
WHERE id = $2 AND embedding IS NULL AND deleted_at IS NULL AND status = 'active'
AND COALESCE(display_name, '') = $3 AND COALESCE(frontmatter, '') = $4`,
vectorToString(embedding), agent.ID, agent.DisplayName, agent.Frontmatter)
if err != nil {
return 0, err
}
return result.RowsAffected()
})
total += updated
if err != nil {
backfillErrs = append(backfillErrs, err)
if ctx.Err() != nil {
return total, errors.Join(backfillErrs...)
}
}
cursor = pending[len(pending)-1].ID
}
slog.Info("agent embeddings backfill complete", "updated", updated)
return updated, nil
}
// agentSelectCols is the column list for all agent SELECT queries.
@@ -155,9 +207,9 @@ func (s *PGAgentStore) Create(ctx context.Context, agent *store.AgentData) error
return fmt.Errorf("commit agent create: %w", err)
}
// Generate embedding for new agent with frontmatter (best-effort, outside tx)
if agent.Frontmatter != "" && s.embProvider != nil {
go s.generateAgentEmbedding(context.Background(), agent.ID, agent.DisplayName, agent.Frontmatter)
// Generate embedding for new agent with frontmatter (best-effort, outside tx).
if agent.Frontmatter != "" {
s.queueAgentEmbedding(agent.ID)
}
return nil
}
@@ -271,6 +323,13 @@ func (s *PGAgentStore) Update(ctx context.Context, id uuid.UUID, updates map[str
}
}
_, frontmatterChanged := updates["frontmatter"]
_, displayNameChanged := updates["display_name"]
if frontmatterChanged || displayNameChanged {
// Never keep a vector built from stale routing metadata. A bounded async
// job or the next startup backfill will fill this NULL.
updates["embedding"] = nil
}
updates["updated_at"] = time.Now()
if store.IsCrossTenant(ctx) {
if err := execMapUpdateWhere(ctx, s.db, "agents", updates, "id = $IDX AND deleted_at IS NULL", id); err != nil {
@@ -291,15 +350,8 @@ func (s *PGAgentStore) Update(ctx context.Context, id uuid.UUID, updates map[str
}
}
// Regenerate embedding when frontmatter changes
if _, hasFrontmatter := updates["frontmatter"]; hasFrontmatter && s.embProvider != nil {
bgCtx := store.WithTenantID(context.Background(), store.TenantIDFromContext(ctx))
go func() {
ag, agErr := s.GetByID(bgCtx, id)
if agErr == nil {
s.generateAgentEmbedding(bgCtx, id, ag.DisplayName, ag.Frontmatter)
}
}()
if frontmatterChanged || displayNameChanged {
s.queueAgentEmbedding(id)
}
return nil
}
+262
View File
@@ -0,0 +1,262 @@
package pg
import (
"context"
"errors"
"fmt"
"log/slog"
"github.com/google/uuid"
"github.com/nextlevelbuilder/goclaw/internal/store"
)
const embeddingBackfillBatchSize = 50
type embeddingBackfillItem[T any] struct {
row T
id uuid.UUID
text string
}
// processEmbeddingBackfillBatch uses an efficient batch request first, then
// isolates individual rows if the provider rejects or truncates the batch. A
// malformed row remains retryable without starving every later row forever.
func processEmbeddingBackfillBatch[T any](
ctx context.Context,
provider store.EmbeddingProvider,
surface string,
items []embeddingBackfillItem[T],
update func(context.Context, T, []float32) (int64, error),
) (int, error) {
texts := make([]string, len(items))
for i, item := range items {
texts[i] = item.text
}
embeddings, batchErr := provider.Embed(ctx, texts)
if batchErr == nil {
batchErr = validateEmbeddingBatch(embeddings, len(items))
}
if batchErr == nil {
return updateEmbeddingBackfillItems(ctx, surface, items, embeddings, update)
}
if ctx.Err() != nil {
return 0, fmt.Errorf("generate %s embeddings: %w", surface, ctx.Err())
}
slog.Warn("embedding backfill batch failed; retrying rows individually",
"surface", surface, "rows", len(items), "error", batchErr)
total := 0
var rowErrs []error
for _, item := range items {
embedding, err := provider.Embed(ctx, []string{item.text})
if err == nil {
err = validateEmbeddingBatch(embedding, 1)
}
if err != nil {
rowErrs = append(rowErrs, fmt.Errorf("generate %s embedding id=%s: %w", surface, item.id, err))
if ctx.Err() != nil {
break
}
continue
}
updated, err := update(ctx, item.row, embedding[0])
if err != nil {
rowErrs = append(rowErrs, fmt.Errorf("update %s embedding id=%s: %w", surface, item.id, err))
continue
}
total += int(updated)
}
return total, errors.Join(rowErrs...)
}
func validateEmbeddingBatch(embeddings [][]float32, expected int) error {
if len(embeddings) != expected {
return fmt.Errorf("got %d vectors for %d inputs", len(embeddings), expected)
}
for i, embedding := range embeddings {
if len(embedding) == 0 {
return fmt.Errorf("empty vector at index %d", i)
}
}
return nil
}
func updateEmbeddingBackfillItems[T any](
ctx context.Context,
surface string,
items []embeddingBackfillItem[T],
embeddings [][]float32,
update func(context.Context, T, []float32) (int64, error),
) (int, error) {
total := 0
var updateErrs []error
for i, item := range items {
updated, err := update(ctx, item.row, embeddings[i])
if err != nil {
updateErrs = append(updateErrs, fmt.Errorf("update %s embedding id=%s: %w", surface, item.id, err))
continue
}
total += int(updated)
}
return total, errors.Join(updateErrs...)
}
// BackfillVaultEmbeddings generates vectors for vault documents that were
// persisted while the embedding provider was unavailable. Documents without a
// summary still get a title/path vector; enrichment can later replace it with a
// richer title/path/summary vector.
func (s *PGVaultStore) BackfillVaultEmbeddings(ctx context.Context) (int, error) {
if s.embProvider == nil {
return 0, fmt.Errorf("no embedding provider configured")
}
type vaultBackfillRow struct {
id uuid.UUID
title string
path string
summary string
}
total := 0
cursor := uuid.Nil
var backfillErrs []error
for {
rows, err := s.db.QueryContext(ctx, `
SELECT id, title, path, COALESCE(summary, '')
FROM vault_documents
WHERE embedding IS NULL AND id > $1
ORDER BY id ASC
LIMIT $2`, cursor, embeddingBackfillBatchSize)
if err != nil {
return total, fmt.Errorf("query vault documents without embeddings: %w", err)
}
pending := make([]vaultBackfillRow, 0, embeddingBackfillBatchSize)
for rows.Next() {
var row vaultBackfillRow
if err := rows.Scan(&row.id, &row.title, &row.path, &row.summary); err != nil {
rows.Close()
return total, fmt.Errorf("scan vault document for embedding backfill: %w", err)
}
pending = append(pending, row)
}
if err := rows.Err(); err != nil {
rows.Close()
return total, fmt.Errorf("iterate vault documents for embedding backfill: %w", err)
}
if err := rows.Close(); err != nil {
return total, fmt.Errorf("close vault embedding backfill rows: %w", err)
}
if len(pending) == 0 {
return total, errors.Join(backfillErrs...)
}
items := make([]embeddingBackfillItem[vaultBackfillRow], len(pending))
for i, doc := range pending {
items[i] = embeddingBackfillItem[vaultBackfillRow]{
row: doc,
id: doc.id,
text: doc.title + " " + doc.path + " " + doc.summary,
}
}
updated, err := processEmbeddingBackfillBatch(ctx, s.embProvider, "vault document", items,
func(ctx context.Context, doc vaultBackfillRow, embedding []float32) (int64, error) {
result, err := s.db.ExecContext(ctx, `
UPDATE vault_documents SET embedding = $1::vector
WHERE id = $2 AND embedding IS NULL
AND title = $3 AND path = $4 AND COALESCE(summary, '') = $5`,
vectorToString(embedding), doc.id, doc.title, doc.path, doc.summary)
if err != nil {
return 0, err
}
return result.RowsAffected()
})
total += updated
if err != nil {
backfillErrs = append(backfillErrs, err)
if ctx.Err() != nil {
return total, errors.Join(backfillErrs...)
}
}
cursor = pending[len(pending)-1].id
}
}
// BackfillEpisodicEmbeddings generates vectors for unexpired summaries that
// were persisted while the embedding provider was unavailable.
func (s *PGEpisodicStore) BackfillEpisodicEmbeddings(ctx context.Context) (int, error) {
if s.embProvider == nil {
return 0, fmt.Errorf("no embedding provider configured")
}
type episodicBackfillRow struct {
id uuid.UUID
summary string
}
total := 0
cursor := uuid.Nil
var backfillErrs []error
for {
rows, err := s.db.QueryContext(ctx, `
SELECT id, summary
FROM episodic_summaries
WHERE embedding IS NULL AND summary != '' AND id > $1
AND (expires_at IS NULL OR expires_at > NOW())
ORDER BY id ASC
LIMIT $2`, cursor, embeddingBackfillBatchSize)
if err != nil {
return total, fmt.Errorf("query episodic summaries without embeddings: %w", err)
}
pending := make([]episodicBackfillRow, 0, embeddingBackfillBatchSize)
for rows.Next() {
var row episodicBackfillRow
if err := rows.Scan(&row.id, &row.summary); err != nil {
rows.Close()
return total, fmt.Errorf("scan episodic summary for embedding backfill: %w", err)
}
pending = append(pending, row)
}
if err := rows.Err(); err != nil {
rows.Close()
return total, fmt.Errorf("iterate episodic summaries for embedding backfill: %w", err)
}
if err := rows.Close(); err != nil {
return total, fmt.Errorf("close episodic embedding backfill rows: %w", err)
}
if len(pending) == 0 {
return total, errors.Join(backfillErrs...)
}
items := make([]embeddingBackfillItem[episodicBackfillRow], len(pending))
for i, summary := range pending {
items[i] = embeddingBackfillItem[episodicBackfillRow]{
row: summary,
id: summary.id,
text: summary.summary,
}
}
updated, err := processEmbeddingBackfillBatch(ctx, s.embProvider, "episodic summary", items,
func(ctx context.Context, summary episodicBackfillRow, embedding []float32) (int64, error) {
result, err := s.db.ExecContext(ctx, `
UPDATE episodic_summaries SET embedding = $1::vector
WHERE id = $2 AND embedding IS NULL AND summary = $3
AND (expires_at IS NULL OR expires_at > NOW())`,
vectorToString(embedding), summary.id, summary.summary)
if err != nil {
return 0, err
}
return result.RowsAffected()
})
total += updated
if err != nil {
backfillErrs = append(backfillErrs, err)
if ctx.Err() != nil {
return total, errors.Join(backfillErrs...)
}
}
cursor = pending[len(pending)-1].id
}
}
+17 -35
View File
@@ -235,8 +235,8 @@ func (t *TtsTool) resolveTenantProvider(ctx context.Context, mgr *tts.Manager, r
// resolveAgentGenericTTSParams reads the per-agent TTSParams generic map from
// the dispatcher-injected AgentAudioSnapshot. Returns nil when no snapshot
// is present or no tts_params are configured. The caller is responsible for
// calling audio.AdaptAgentParams(generic, providerName) PER-ATTEMPT to convert
// generic keys to provider-specific keys (Finding #1 CRITICAL).
// calling audio.AdaptAgentParams(generic, providerName) for each provider attempt
// so generic keys are converted to the correct provider-specific keys.
func (t *TtsTool) resolveAgentGenericTTSParams(ctx context.Context) map[string]any {
snap, ok := store.AgentAudioFromCtx(ctx)
if !ok || len(snap.OtherConfig) == 0 {
@@ -275,7 +275,7 @@ func (t *TtsTool) Execute(ctx context.Context, args map[string]any) *Result {
argModel, _ := args["model"].(string)
providerName, _ := args["provider"].(string)
// Read generic agent TTS params once; adapt PER-ATTEMPT below (Finding #1 CRITICAL).
// Read generic agent TTS params once; adapt them for each provider attempt below.
// Storing generic keys here so each fallback provider gets its own adapted copy.
genericAgentParams := t.resolveAgentGenericTTSParams(ctx)
@@ -284,17 +284,16 @@ func (t *TtsTool) Execute(ctx context.Context, args map[string]any) *Result {
mgr := t.manager
t.mu.RUnlock()
effectiveProvider := providerName
if effectiveProvider == "" {
effectiveProvider = t.resolvePrimary(ctx, mgr)
}
voice, model, voiceFromAgent := t.resolveVoiceAndModel(ctx, effectiveProvider, argVoice, argModel)
// Determine format based on channel (read from ctx — thread-safe)
channel := ToolChannelFromCtx(ctx)
opts := tts.Options{Voice: voice, Model: model}
if channel == "telegram" {
opts.Format = "opus"
resolveProviderOpts := func(name string) tts.Options {
voice, model, voiceFromAgent := t.resolveVoiceAndModel(ctx, name, argVoice, argModel)
opts := tts.Options{Voice: voice, Model: model}
if channel == "telegram" {
opts.Format = "opus"
}
opts.Voice = applyVoiceCompat(name, opts.Voice, voiceFromAgent)
return opts
}
var result *tts.SynthResult
@@ -312,7 +311,7 @@ func (t *TtsTool) Execute(ctx context.Context, args map[string]any) *Result {
}
providerName = tenantName
}
opts.Voice = applyVoiceCompat(providerName, opts.Voice, voiceFromAgent)
opts := resolveProviderOpts(providerName)
if adapted := audio.AdaptAgentParams(genericAgentParams, providerName); len(adapted) > 0 {
opts.Params = mergeParams(opts.Params, adapted)
}
@@ -320,37 +319,20 @@ func (t *TtsTool) Execute(ctx context.Context, args map[string]any) *Result {
} else {
// Prefer the DB-configured tenant provider used by /tts and auto-TTS.
if p, tenantName, ok := t.resolveTenantProvider(ctx, mgr, ""); ok {
tenantOpts := opts
tenantOpts.Voice = applyVoiceCompat(tenantName, tenantOpts.Voice, voiceFromAgent)
tenantOpts := resolveProviderOpts(tenantName)
if adapted := audio.AdaptAgentParams(genericAgentParams, tenantName); len(adapted) > 0 {
tenantOpts.Params = mergeParams(opts.Params, adapted)
tenantOpts.Params = mergeParams(tenantOpts.Params, adapted)
}
result, err = p.Synthesize(ctx, text, tenantOpts)
if err != nil {
slog.Warn("tts tenant provider failed, trying fallback", "provider", tenantName, "error", err)
result, err = mgr.SynthesizeWithFallbackAdapted(ctx, text, opts, genericAgentParams)
primary := t.resolvePrimary(ctx, mgr)
result, err = mgr.SynthesizeWithFallbackResolved(ctx, text, primary, resolveProviderOpts, genericAgentParams)
}
} else {
// Resolve primary from tenant settings or default.
primary := t.resolvePrimary(ctx, mgr)
if p, ok := mgr.GetProvider(primary); ok {
// Adapt for the primary provider attempt specifically.
primaryOpts := opts
primaryOpts.Voice = applyVoiceCompat(primary, primaryOpts.Voice, voiceFromAgent)
if adapted := audio.AdaptAgentParams(genericAgentParams, primary); len(adapted) > 0 {
primaryOpts.Params = mergeParams(opts.Params, adapted)
}
result, err = p.Synthesize(ctx, text, primaryOpts)
if err != nil {
slog.Warn("tts primary provider failed, trying fallback", "provider", primary, "error", err)
// SynthesizeWithFallbackAdapted adapts genericAgentParams per-attempt
// (Finding #1 CRITICAL): each fallback provider receives its own
// provider-native keys, not the primary's adapted map.
result, err = mgr.SynthesizeWithFallbackAdapted(ctx, text, opts, genericAgentParams)
}
} else {
result, err = mgr.SynthesizeWithFallbackAdapted(ctx, text, opts, genericAgentParams)
}
result, err = mgr.SynthesizeWithFallbackResolved(ctx, text, primary, resolveProviderOpts, genericAgentParams)
}
}
+94 -2
View File
@@ -14,18 +14,57 @@ type tenantRoutingTTSProvider struct {
name string
calls int
err error
voice string
model string
}
func (p *tenantRoutingTTSProvider) Name() string { return p.name }
func (p *tenantRoutingTTSProvider) Synthesize(_ context.Context, _ string, _ tts.Options) (*tts.SynthResult, error) {
func (p *tenantRoutingTTSProvider) Synthesize(_ context.Context, _ string, opts tts.Options) (*tts.SynthResult, error) {
p.calls++
p.voice = opts.Voice
p.model = opts.Model
if p.err != nil {
return nil, p.err
}
return &tts.SynthResult{Audio: []byte("audio"), Extension: "mp3", MimeType: "audio/mpeg"}, nil
}
func TestTtsTool_UsesSystemConfigForResolvedTenantProvider(t *testing.T) {
t.Parallel()
globalProvider := &tenantRoutingTTSProvider{name: "openai"}
tenantProvider := &tenantRoutingTTSProvider{name: "edge"}
mgr := newTenantRoutingManager("openai", globalProvider)
mgr.SetTenantResolver(func(context.Context) (audio.TTSProvider, string, audio.AutoMode, error) {
return tenantProvider, "edge", audio.AutoOff, nil
})
tool := NewTtsTool(mgr)
tool.SetSystemConfigStore(&fakeSystemConfigStore{data: map[string]string{
"tts.openai.voice": "alloy",
"tts.openai.model": "openai-model",
"tts.edge.voice": "vi-VN-HoaiMyNeural",
"tts.edge.model": "edge-model",
}})
result := tool.Execute(WithToolWorkspace(context.Background(), t.TempDir()), map[string]any{
"text": "hello from tenant provider",
})
if result.IsError {
t.Fatalf("unexpected error: %s", result.ForLLM)
}
if tenantProvider.voice != "vi-VN-HoaiMyNeural" {
t.Fatalf("tenant voice = %q, want %q", tenantProvider.voice, "vi-VN-HoaiMyNeural")
}
if tenantProvider.model != "edge-model" {
t.Fatalf("tenant model = %q, want %q", tenantProvider.model, "edge-model")
}
if globalProvider.calls != 0 {
t.Fatalf("global provider calls = %d, want 0", globalProvider.calls)
}
}
func newTenantRoutingManager(primary string, providers ...*tenantRoutingTTSProvider) *tts.Manager {
mgr := tts.NewManager(tts.ManagerConfig{Primary: primary})
for _, provider := range providers {
@@ -98,7 +137,13 @@ func TestTtsTool_TenantProviderFailureFallsBackWhenProviderOmitted(t *testing.T)
})
tool := NewTtsTool(mgr)
result := tool.Execute(context.Background(), map[string]any{
tool.SetSystemConfigStore(&fakeSystemConfigStore{data: map[string]string{
"tts.edge.voice": "vi-VN-HoaiMyNeural",
"tts.edge.model": "edge-model",
"tts.gemini.voice": "Kore",
"tts.gemini.model": "gemini-model",
}})
result := tool.Execute(WithToolWorkspace(context.Background(), t.TempDir()), map[string]any{
"text": "hello with tenant fallback",
})
@@ -111,6 +156,53 @@ func TestTtsTool_TenantProviderFailureFallsBackWhenProviderOmitted(t *testing.T)
if edgeProvider.calls != 1 {
t.Fatalf("global edge calls = %d, want 1", edgeProvider.calls)
}
if geminiProvider.voice != "Kore" || geminiProvider.model != "gemini-model" {
t.Fatalf("tenant options = voice %q model %q, want tenant system config", geminiProvider.voice, geminiProvider.model)
}
if edgeProvider.voice != "vi-VN-HoaiMyNeural" || edgeProvider.model != "edge-model" {
t.Fatalf("fallback options = voice %q model %q, want global system config", edgeProvider.voice, edgeProvider.model)
}
}
func TestTtsTool_FallbackResolvesOptionsForEveryActualProvider(t *testing.T) {
t.Parallel()
managerPrimary := &tenantRoutingTTSProvider{name: "edge"}
preferred := &tenantRoutingTTSProvider{name: "openai", err: errors.New("openai unavailable")}
tenantProvider := &tenantRoutingTTSProvider{name: "gemini", err: errors.New("tenant provider unavailable")}
mgr := newTenantRoutingManager("edge", managerPrimary, preferred)
mgr.SetTenantResolver(func(context.Context) (audio.TTSProvider, string, audio.AutoMode, error) {
return tenantProvider, "gemini", audio.AutoOff, nil
})
tool := NewTtsTool(mgr)
tool.SetSystemConfigStore(&fakeSystemConfigStore{data: map[string]string{
"tts.gemini.voice": "Kore",
"tts.gemini.model": "gemini-model",
"tts.openai.voice": "alloy",
"tts.openai.model": "openai-model",
"tts.edge.voice": "vi-VN-HoaiMyNeural",
"tts.edge.model": "edge-model",
}})
ctx := ctxWithTTSSettings(t, ttsOverride{Primary: "openai"})
ctx = WithToolWorkspace(ctx, t.TempDir())
result := tool.Execute(ctx, map[string]any{"text": "fallback across providers"})
if result.IsError {
t.Fatalf("unexpected error: %s", result.ForLLM)
}
if tenantProvider.voice != "Kore" || tenantProvider.model != "gemini-model" {
t.Fatalf("tenant options = voice %q model %q, want gemini config", tenantProvider.voice, tenantProvider.model)
}
if preferred.voice != "alloy" || preferred.model != "openai-model" {
t.Fatalf("preferred options = voice %q model %q, want openai config", preferred.voice, preferred.model)
}
if managerPrimary.voice != "vi-VN-HoaiMyNeural" || managerPrimary.model != "edge-model" {
t.Fatalf("secondary options = voice %q model %q, want edge config", managerPrimary.voice, managerPrimary.model)
}
if preferred.calls != 1 || managerPrimary.calls != 1 {
t.Fatalf("fallback calls = preferred %d secondary %d, want one each", preferred.calls, managerPrimary.calls)
}
}
func TestTtsTool_ExplicitProviderMismatchStillErrors(t *testing.T) {
@@ -0,0 +1,341 @@
//go:build integration
package integration
import (
"context"
"database/sql"
"errors"
"strings"
"sync"
"testing"
"time"
"github.com/google/uuid"
"github.com/nextlevelbuilder/goclaw/internal/store"
"github.com/nextlevelbuilder/goclaw/internal/store/pg"
)
func TestStoreVault_BackfillEmbeddings_Idempotent(t *testing.T) {
db := testDB(t)
tenantID, agentID := seedTenantAgent(t, db)
ctx := tenantCtx(tenantID)
storeWithoutProvider := pg.NewPGVaultStore(db)
docWithSummary := makeVaultDoc(tenantID.String(), agentID.String(), "backfill/summary.md", "Recovery summary")
docWithSummary.Summary = "Semantic recovery content"
if err := storeWithoutProvider.UpsertDocument(ctx, docWithSummary); err != nil {
t.Fatalf("UpsertDocument with summary: %v", err)
}
docWithoutSummary := makeVaultDoc(tenantID.String(), agentID.String(), "backfill/path-only.md", "Recovery path")
if err := storeWithoutProvider.UpsertDocument(ctx, docWithoutSummary); err != nil {
t.Fatalf("UpsertDocument without summary: %v", err)
}
storeWithoutProvider.SetEmbeddingProvider(newMockEmbedProvider())
updated, err := storeWithoutProvider.BackfillVaultEmbeddings(ctx)
if err != nil {
t.Fatalf("BackfillVaultEmbeddings: %v", err)
}
if updated != 2 {
t.Fatalf("BackfillVaultEmbeddings updated = %d, want 2", updated)
}
assertEmbeddingPresent(t, db, "SELECT embedding IS NOT NULL FROM vault_documents WHERE id = $1", "vault document", docWithSummary.ID)
assertEmbeddingPresent(t, db, "SELECT embedding IS NOT NULL FROM vault_documents WHERE id = $1", "vault document", docWithoutSummary.ID)
updated, err = storeWithoutProvider.BackfillVaultEmbeddings(ctx)
if err != nil {
t.Fatalf("BackfillVaultEmbeddings second call: %v", err)
}
if updated != 0 {
t.Fatalf("BackfillVaultEmbeddings second call updated = %d, want 0", updated)
}
}
func TestStoreEpisodic_BackfillEmbeddings_Idempotent(t *testing.T) {
db := testDB(t)
tenantID, agentID := seedTenantAgent(t, db)
ctx := tenantCtx(tenantID)
storeWithoutProvider := pg.NewPGEpisodicStore(db)
episode := &store.EpisodicSummary{
TenantID: tenantID,
AgentID: agentID,
UserID: "embedding-backfill-user",
SessionKey: "embedding-backfill-session",
Summary: "The user decided to restore all missing semantic vectors.",
SourceType: "session",
SourceID: "embedding-backfill-" + uuid.NewString(),
}
if err := storeWithoutProvider.Create(ctx, episode); err != nil {
t.Fatalf("Create episodic summary: %v", err)
}
expiredAt := time.Now().Add(-time.Hour)
expiredEpisode := &store.EpisodicSummary{
TenantID: tenantID,
AgentID: agentID,
UserID: "embedding-backfill-user",
SessionKey: "expired-embedding-backfill-session",
Summary: "This expired summary must not consume provider capacity.",
SourceType: "session",
SourceID: "expired-embedding-backfill-" + uuid.NewString(),
ExpiresAt: &expiredAt,
}
if err := storeWithoutProvider.Create(ctx, expiredEpisode); err != nil {
t.Fatalf("Create expired episodic summary: %v", err)
}
storeWithoutProvider.SetEmbeddingProvider(newMockEmbedProvider())
updated, err := storeWithoutProvider.BackfillEpisodicEmbeddings(ctx)
if err != nil {
t.Fatalf("BackfillEpisodicEmbeddings: %v", err)
}
if updated != 1 {
t.Fatalf("BackfillEpisodicEmbeddings updated = %d, want 1", updated)
}
assertEmbeddingPresent(t, db, "SELECT embedding IS NOT NULL FROM episodic_summaries WHERE id = $1", "episodic summary", episode.ID.String())
assertEmbeddingMissing(t, db, "SELECT embedding IS NULL FROM episodic_summaries WHERE id = $1", "expired episodic summary", expiredEpisode.ID.String())
updated, err = storeWithoutProvider.BackfillEpisodicEmbeddings(ctx)
if err != nil {
t.Fatalf("BackfillEpisodicEmbeddings second call: %v", err)
}
if updated != 0 {
t.Fatalf("BackfillEpisodicEmbeddings second call updated = %d, want 0", updated)
}
}
func TestStoreAgent_BackfillEmbeddings_Idempotent(t *testing.T) {
db := testDB(t)
tenantID, agentID := seedTenantAgent(t, db)
ctx := tenantCtx(tenantID)
agentStore := pg.NewPGAgentStore(db)
if err := agentStore.Update(ctx, agentID, map[string]any{
"display_name": "Recovery agent",
"frontmatter": "An agent with semantic routing metadata",
}); err != nil {
t.Fatalf("Update agent frontmatter: %v", err)
}
agentStore.SetEmbeddingProvider(newMockEmbedProvider())
updated, err := agentStore.BackfillAgentEmbeddings(ctx)
if err != nil {
t.Fatalf("BackfillAgentEmbeddings: %v", err)
}
if updated != 1 {
t.Fatalf("BackfillAgentEmbeddings updated = %d, want 1", updated)
}
assertEmbeddingPresent(t, db, "SELECT embedding IS NOT NULL FROM agents WHERE id = $1", "agent", agentID.String())
updated, err = agentStore.BackfillAgentEmbeddings(ctx)
if err != nil {
t.Fatalf("BackfillAgentEmbeddings second call: %v", err)
}
if updated != 0 {
t.Fatalf("BackfillAgentEmbeddings second call updated = %d, want 0", updated)
}
}
func TestStoreAgent_AsyncEmbeddingDoesNotOverwriteNewerSource(t *testing.T) {
db := testDB(t)
tenantID, agentID := seedTenantAgent(t, db)
ctx := tenantCtx(tenantID)
provider := newBlockingAgentEmbedProvider()
agentStore := pg.NewPGAgentStore(db)
agentStore.SetEmbeddingProvider(provider)
if err := agentStore.Update(ctx, agentID, map[string]any{
"display_name": "Versioned agent",
"frontmatter": "old routing metadata",
}); err != nil {
t.Fatalf("Update old agent source: %v", err)
}
select {
case <-provider.oldStarted:
case <-time.After(3 * time.Second):
t.Fatal("old embedding request did not start")
}
if err := agentStore.Update(ctx, agentID, map[string]any{
"frontmatter": "new routing metadata",
}); err != nil {
t.Fatalf("Update new agent source: %v", err)
}
assertAgentEmbeddingEquals(t, db, agentID, vectorLiteral(0.2))
close(provider.releaseOld)
select {
case <-provider.oldDone:
case <-time.After(3 * time.Second):
t.Fatal("old embedding request did not finish")
}
assertAgentEmbeddingEquals(t, db, agentID, vectorLiteral(0.2))
}
func TestStoreVault_BackfillEmbeddings_RejectsIncompleteBatch(t *testing.T) {
db := testDB(t)
tenantID, agentID := seedTenantAgent(t, db)
ctx := tenantCtx(tenantID)
vaultStore := pg.NewPGVaultStore(db)
doc := makeVaultDoc(tenantID.String(), agentID.String(), "backfill/incomplete.md", "Incomplete response")
doc.Summary = "This row must remain pending when the provider drops a vector."
if err := vaultStore.UpsertDocument(ctx, doc); err != nil {
t.Fatalf("UpsertDocument: %v", err)
}
vaultStore.SetEmbeddingProvider(incompleteEmbedProvider{})
if _, err := vaultStore.BackfillVaultEmbeddings(ctx); err == nil {
t.Fatal("BackfillVaultEmbeddings error = nil, want incomplete batch error")
}
assertEmbeddingMissing(t, db, "SELECT embedding IS NULL FROM vault_documents WHERE id = $1", "vault document", doc.ID)
}
func TestStoreVault_BackfillEmbeddings_PoisonRowDoesNotStarveLaterRows(t *testing.T) {
db := testDB(t)
tenantID, agentID := seedTenantAgent(t, db)
ctx := tenantCtx(tenantID)
vaultStore := pg.NewPGVaultStore(db)
poison := makeVaultDoc(tenantID.String(), agentID.String(), "backfill/poison.md", "Poison row")
poison.Summary = "poison embedding input"
valid := makeVaultDoc(tenantID.String(), agentID.String(), "backfill/valid.md", "Valid row")
valid.Summary = "valid embedding input"
if err := vaultStore.UpsertDocument(ctx, poison); err != nil {
t.Fatalf("Upsert poison document: %v", err)
}
if err := vaultStore.UpsertDocument(ctx, valid); err != nil {
t.Fatalf("Upsert valid document: %v", err)
}
vaultStore.SetEmbeddingProvider(poisonEmbedProvider{delegate: newMockEmbedProvider()})
updated, err := vaultStore.BackfillVaultEmbeddings(ctx)
if err == nil {
t.Fatal("BackfillVaultEmbeddings error = nil, want poison-row error")
}
if updated != 1 {
t.Fatalf("BackfillVaultEmbeddings updated = %d, want 1 valid row", updated)
}
assertEmbeddingMissing(t, db, "SELECT embedding IS NULL FROM vault_documents WHERE id = $1", "poison vault document", poison.ID)
assertEmbeddingPresent(t, db, "SELECT embedding IS NOT NULL FROM vault_documents WHERE id = $1", "valid vault document", valid.ID)
}
func assertEmbeddingPresent(t *testing.T, db queryRower, query, surface, id string) {
t.Helper()
var present bool
if err := db.QueryRow(query, id).Scan(&present); err != nil {
t.Fatalf("query %s embedding: %v", surface, err)
}
if !present {
t.Fatalf("expected %s embedding to be present", surface)
}
}
func assertEmbeddingMissing(t *testing.T, db queryRower, query, surface, id string) {
t.Helper()
var missing bool
if err := db.QueryRow(query, id).Scan(&missing); err != nil {
t.Fatalf("query %s embedding: %v", surface, err)
}
if !missing {
t.Fatalf("expected %s embedding to remain missing", surface)
}
}
type queryRower interface {
QueryRow(query string, args ...any) *sql.Row
}
type incompleteEmbedProvider struct{}
func (incompleteEmbedProvider) Name() string { return "incomplete" }
func (incompleteEmbedProvider) Model() string { return "incomplete" }
func (incompleteEmbedProvider) Embed(context.Context, []string) ([][]float32, error) {
return nil, nil
}
var _ store.EmbeddingProvider = incompleteEmbedProvider{}
type poisonEmbedProvider struct {
delegate store.EmbeddingProvider
}
func (p poisonEmbedProvider) Name() string { return "poison" }
func (p poisonEmbedProvider) Model() string { return "poison" }
func (p poisonEmbedProvider) Embed(ctx context.Context, texts []string) ([][]float32, error) {
for _, text := range texts {
if strings.Contains(text, "poison") {
return nil, errors.New("rejected poison input")
}
}
return p.delegate.Embed(ctx, texts)
}
var _ store.EmbeddingProvider = poisonEmbedProvider{}
type blockingAgentEmbedProvider struct {
oldStarted chan struct{}
releaseOld chan struct{}
oldDone chan struct{}
startOnce sync.Once
doneOnce sync.Once
}
func newBlockingAgentEmbedProvider() *blockingAgentEmbedProvider {
return &blockingAgentEmbedProvider{
oldStarted: make(chan struct{}),
releaseOld: make(chan struct{}),
oldDone: make(chan struct{}),
}
}
func (p *blockingAgentEmbedProvider) Name() string { return "blocking-agent" }
func (p *blockingAgentEmbedProvider) Model() string { return "blocking-agent" }
func (p *blockingAgentEmbedProvider) Embed(ctx context.Context, texts []string) ([][]float32, error) {
value := float32(0.2)
if strings.Contains(texts[0], "old routing metadata") {
value = 0.1
p.startOnce.Do(func() { close(p.oldStarted) })
select {
case <-p.releaseOld:
case <-ctx.Done():
return nil, ctx.Err()
}
p.doneOnce.Do(func() { close(p.oldDone) })
}
result := make([][]float32, len(texts))
for i := range result {
result[i] = make([]float32, store.RequiredMemoryEmbeddingDimensions)
result[i][0] = value
}
return result, nil
}
func assertAgentEmbeddingEquals(t *testing.T, db queryRower, agentID uuid.UUID, expected string) {
t.Helper()
deadline := time.Now().Add(3 * time.Second)
for time.Now().Before(deadline) {
var equal bool
if err := db.QueryRow(`SELECT COALESCE(embedding = $1::vector, false) FROM agents WHERE id = $2`, expected, agentID).Scan(&equal); err != nil {
t.Fatalf("query agent embedding: %v", err)
}
if equal {
return
}
time.Sleep(10 * time.Millisecond)
}
t.Fatal("agent embedding did not match the newest source")
}
func vectorLiteral(first float64) string {
values := make([]string, store.RequiredMemoryEmbeddingDimensions)
values[0] = "0.2"
if first == 0.1 {
values[0] = "0.1"
}
for i := 1; i < len(values); i++ {
values[i] = "0"
}
return "[" + strings.Join(values, ",") + "]"
}
var _ store.EmbeddingProvider = (*blockingAgentEmbedProvider)(nil)
+4 -1
View File
@@ -72,6 +72,7 @@
"headers": "Headers",
"bodyTemplate": "Body Template",
"promptTemplate": "Prompt Template",
"provider": "Provider",
"model": "Model",
"maxInvocationsPerTurn": "Max Invocations per Turn",
"scriptSource": "Script source (ES5.1 JavaScript)",
@@ -144,6 +145,8 @@
"promptRequiresMatcher": "Prompt hooks require a matcher or if_expr",
"invalidRegex": "Invalid regex",
"promptTemplateRequired": "Prompt template is required",
"promptProviderRequired": "Provider is required",
"promptModelRequired": "Model is required",
"scriptSourceRequired": "Script source is required"
},
"decision": {
@@ -152,4 +155,4 @@
"error": "Error",
"timeout": "Timeout"
}
}
}
+3
View File
@@ -72,6 +72,7 @@
"headers": "Заголовки",
"bodyTemplate": "Шаблон тела",
"promptTemplate": "Шаблон промпта",
"provider": "Провайдер",
"model": "Модель",
"maxInvocationsPerTurn": "Макс. вызовов за ход",
"scriptSource": "Исходный код скрипта (ES5.1 JavaScript)",
@@ -138,6 +139,8 @@
"promptRequiresMatcher": "Хуки prompt требуют матчер или if_expr",
"invalidRegex": "Недопустимый regex",
"promptTemplateRequired": "Шаблон промпта обязателен",
"promptProviderRequired": "Провайдер обязателен",
"promptModelRequired": "Модель обязательна",
"scriptSourceRequired": "Исходный код скрипта обязателен"
},
"decision": {
+4 -1
View File
@@ -72,6 +72,7 @@
"headers": "Headers",
"bodyTemplate": "Mẫu body",
"promptTemplate": "Mẫu prompt",
"provider": "Nhà cung cấp",
"model": "Mô hình",
"maxInvocationsPerTurn": "Số lần gọi tối đa mỗi lượt",
"scriptSource": "Mã nguồn script (ES5.1 JavaScript)",
@@ -144,6 +145,8 @@
"promptRequiresMatcher": "Hook prompt yêu cầu matcher hoặc if_expr",
"invalidRegex": "Regex không hợp lệ",
"promptTemplateRequired": "Mẫu prompt là bắt buộc",
"promptProviderRequired": "Nhà cung cấp là bắt buộc",
"promptModelRequired": "Mô hình là bắt buộc",
"scriptSourceRequired": "Mã nguồn script là bắt buộc"
},
"decision": {
@@ -152,4 +155,4 @@
"error": "Lỗi",
"timeout": "Hết thời gian"
}
}
}
+4 -1
View File
@@ -72,6 +72,7 @@
"headers": "请求头",
"bodyTemplate": "请求体模板",
"promptTemplate": "Prompt 模板",
"provider": "提供商",
"model": "模型",
"maxInvocationsPerTurn": "每轮最大调用次数",
"scriptSource": "脚本源码(ES5.1 JavaScript)",
@@ -144,6 +145,8 @@
"promptRequiresMatcher": "Prompt 钩子需要 matcher 或 if_expr",
"invalidRegex": "无效的正则表达式",
"promptTemplateRequired": "Prompt 模板为必填项",
"promptProviderRequired": "提供商为必填项",
"promptModelRequired": "模型为必填项",
"scriptSourceRequired": "脚本源码为必填项"
},
"decision": {
@@ -152,4 +155,4 @@
"error": "错误",
"timeout": "超时"
}
}
}
@@ -15,6 +15,7 @@ import {
import { HookListRow } from "@/pages/hooks/components/hook-list-row";
import { HookFormDialog } from "@/pages/hooks/components/hook-form-dialog";
import { HookTestPanel } from "@/pages/hooks/components/hook-test-panel";
import { buildHookConfig } from "@/pages/hooks/hook-form-config";
import type { HookFormData } from "@/schemas/hooks.schema";
interface AgentHooksTabProps {
@@ -23,42 +24,6 @@ interface AgentHooksTabProps {
onCreateOpenChange?: (open: boolean) => void;
}
function parseHeaders(raw: string | undefined): Record<string, unknown> {
const trimmed = (raw ?? "").trim();
if (!trimmed) return {};
try {
const parsed = JSON.parse(trimmed);
if (parsed && typeof parsed === "object" && !Array.isArray(parsed)) {
return parsed as Record<string, unknown>;
}
throw new Error("headers must be a JSON object");
} catch (err) {
// eslint-disable-next-line preserve-caught-error -- JSON.parse error message already captured verbatim in thrown message
throw new Error(
"Invalid headers JSON: " + (err instanceof Error ? err.message : String(err)),
);
}
}
function buildConfig(data: HookFormData): Record<string, unknown> {
if (data.handler_type === "http") {
return {
url: data.url ?? "",
method: data.method ?? "POST",
headers: parseHeaders(data.headers),
body_template: data.body_template ?? "",
};
}
if (data.handler_type === "script") {
return { source: data.script_source ?? "" };
}
return {
prompt_template: data.prompt_template ?? "",
model: data.model ?? "haiku",
max_invocations_per_turn: data.max_invocations_per_turn ?? 5,
};
}
export function AgentHooksTab({ agentId, initialCreateOpen, onCreateOpenChange }: AgentHooksTabProps) {
const { t } = useTranslation("agents");
const { t: th } = useTranslation("hooks");
@@ -89,7 +54,7 @@ export function AgentHooksTab({ agentId, initialCreateOpen, onCreateOpenChange }
const handleCreate = async (data: HookFormData) => {
let config: Record<string, unknown>;
try {
config = buildConfig(data);
config = buildHookConfig(data);
} catch (err) {
toast.error(err instanceof Error ? err.message : String(err));
return;
@@ -118,7 +83,7 @@ export function AgentHooksTab({ agentId, initialCreateOpen, onCreateOpenChange }
if (!editTarget) return;
let config: Record<string, unknown>;
try {
config = buildConfig(data);
config = buildHookConfig(data);
} catch (err) {
toast.error(err instanceof Error ? err.message : String(err));
return;
@@ -6,6 +6,7 @@
*/
import { describe, it, expect } from "vitest";
import { hookFormSchema } from "@/schemas/hooks.schema";
import { buildHookConfig } from "@/pages/hooks/hook-form-config";
// --- Zod schema validation ---
@@ -33,6 +34,8 @@ describe("hookFormSchema — base cases", () => {
handler_type: "prompt",
matcher: "^bash$",
prompt_template: "Evaluate the tool call.",
provider: "cppai",
model: "gpt-5.6-terra",
});
expect(result.success).toBe(true);
});
@@ -42,6 +45,8 @@ describe("hookFormSchema — base cases", () => {
...base,
handler_type: "prompt",
prompt_template: "Evaluate the tool call.",
provider: "cppai",
model: "gpt-5.6-terra",
matcher: "",
if_expr: "",
});
@@ -57,6 +62,8 @@ describe("hookFormSchema — base cases", () => {
...base,
handler_type: "prompt",
matcher: "^bash$",
provider: "cppai",
model: "gpt-5.6-terra",
});
expect(result.success).toBe(false);
if (!result.success) {
@@ -65,6 +72,21 @@ describe("hookFormSchema — base cases", () => {
}
});
it("rejects prompt hook without provider or model", () => {
const result = hookFormSchema.safeParse({
...base,
handler_type: "prompt",
matcher: "^bash$",
prompt_template: "Evaluate the tool call.",
});
expect(result.success).toBe(false);
if (!result.success) {
const paths = result.error.issues.map((e) => e.path.join("."));
expect(paths).toContain("provider");
expect(paths).toContain("model");
}
});
it("rejects invalid regex in matcher", () => {
const result = hookFormSchema.safeParse({
...base,
@@ -132,6 +154,17 @@ describe("hooks i18n key contracts", () => {
}
});
it("prompt provider/model keys exist in all locales", async () => {
for (const locale of ["en", "vi", "zh", "ru"]) {
const mod = await import(`@/i18n/locales/${locale}/hooks.json`);
const data = mod as unknown as Record<string, Record<string, string>>;
expect(data.form?.provider).toBeTruthy();
expect(data.form?.model).toBeTruthy();
expect(data.validation?.promptProviderRequired).toBeTruthy();
expect(data.validation?.promptModelRequired).toBeTruthy();
}
});
it("toast keys exist in en", async () => {
const en = await import("@/i18n/locales/en/hooks.json");
const data = en as unknown as Record<string, Record<string, string>>;
@@ -144,14 +177,24 @@ describe("hooks i18n key contracts", () => {
// --- buildConfig helper (re-tested inline) ---
describe("hook config builder logic", () => {
it("prompt type config has prompt_template and model", () => {
const config = {
it("prompt type config persists provider and model", () => {
const config = buildHookConfig({
event: "pre_tool_use",
handler_type: "prompt",
scope: "tenant",
matcher: "^exec$",
timeout_ms: 5000,
on_timeout: "block",
priority: 100,
enabled: true,
prompt_template: "Evaluate this.",
model: "haiku",
provider: "cppai",
model: "gpt-5.6-terra",
max_invocations_per_turn: 5,
};
});
expect(config.prompt_template).toBeTruthy();
expect(config.model).toBe("haiku");
expect(config.provider).toBe("cppai");
expect(config.model).toBe("gpt-5.6-terra");
expect(config.max_invocations_per_turn).toBe(5);
});
@@ -16,6 +16,7 @@ import { hookFormSchema, type HookFormData } from "@/schemas/hooks.schema";
import type { HookConfig } from "@/hooks/use-hooks";
import { useAuthStore } from "@/stores/use-auth-store";
import { useAgents } from "@/pages/agents/hooks/use-agents";
import { ProviderModelSelect } from "@/components/shared/provider-model-select";
import { ScriptEditor } from "./script-editor";
const HOOK_EVENTS = [
@@ -39,7 +40,7 @@ export function HookFormDialog({ open, onOpenChange, onSubmit, initial }: HookFo
: (["tenant", "agent"] as const);
const {
register, control, handleSubmit, watch, reset,
register, control, handleSubmit, watch, reset, setValue,
formState: { errors, isSubmitting },
} = useForm<HookFormData>({
resolver: zodResolver(hookFormSchema),
@@ -54,12 +55,16 @@ export function HookFormDialog({ open, onOpenChange, onSubmit, initial }: HookFo
priority: 100,
enabled: true,
method: "POST",
provider: "",
model: "",
max_invocations_per_turn: 5,
},
});
const handlerType = watch("handler_type");
const scope = watch("scope");
const promptProvider = watch("provider") ?? "";
const promptModel = watch("model") ?? "";
const { agents } = useAgents();
// Builtin rows (Phase 04/05) ship with source='builtin'. UI + backend agree:
// only `enabled` is mutable. All other inputs render as read-only, and the
@@ -94,6 +99,7 @@ export function HookFormDialog({ open, onOpenChange, onSubmit, initial }: HookFo
headers: cfg.headers ? JSON.stringify(cfg.headers) : "",
body_template: (cfg.body_template as string) ?? "",
prompt_template: (cfg.prompt_template as string) ?? "",
provider: (cfg.provider as string) ?? "",
model: (cfg.model as string) ?? "",
max_invocations_per_turn: (cfg.max_invocations_per_turn as number) ?? 5,
script_source: (cfg.source as string) ?? "",
@@ -326,19 +332,29 @@ export function HookFormDialog({ open, onOpenChange, onSubmit, initial }: HookFo
)}
</div>
<div className="space-y-1.5">
<Label>{t("form.model")}</Label>
<Controller control={control} name="model" render={({ field }) => (
<Select value={field.value ?? "haiku"} onValueChange={field.onChange} disabled={isBuiltin}>
<SelectTrigger className="text-base md:text-sm">
<SelectValue />
</SelectTrigger>
<SelectContent>
<SelectItem value="haiku">haiku</SelectItem>
<SelectItem value="sonnet">sonnet</SelectItem>
<SelectItem value="opus">opus</SelectItem>
</SelectContent>
</Select>
)} />
<ProviderModelSelect
provider={promptProvider}
onProviderChange={(value) => {
setValue("provider", value, { shouldDirty: true, shouldValidate: true });
setValue("model", "", { shouldDirty: true, shouldValidate: true });
}}
model={promptModel}
onModelChange={(value) => setValue("model", value, { shouldDirty: true, shouldValidate: true })}
providerLabel={t("form.provider")}
modelLabel={t("form.model")}
showVerify
disabled={isBuiltin}
/>
{errors.provider && (
<p className="text-xs text-destructive">
{t(errors.provider.message ?? "validation.promptProviderRequired")}
</p>
)}
{errors.model && (
<p className="text-xs text-destructive">
{t(errors.model.message ?? "validation.promptModelRequired")}
</p>
)}
</div>
<div className="space-y-1.5">
<Label>{t("form.maxInvocationsPerTurn")}</Label>
@@ -163,6 +163,7 @@ function HttpConfigCard({ cfg, t }: { cfg: Record<string, unknown>; t: TFn }) {
function PromptConfigCard({ cfg, t }: { cfg: Record<string, unknown>; t: TFn }) {
const promptTemplate = typeof cfg.prompt_template === "string" ? cfg.prompt_template : "";
const provider = typeof cfg.provider === "string" ? cfg.provider : "";
const model = typeof cfg.model === "string" ? cfg.model : "";
return (
<div className="rounded-lg border bg-card p-4 space-y-3">
@@ -173,10 +174,17 @@ function PromptConfigCard({ cfg, t }: { cfg: Record<string, unknown>; t: TFn })
<pre className="overflow-x-auto rounded bg-muted px-2 py-1 text-xs whitespace-pre-wrap">{promptTemplate}</pre>
</div>
)}
{model && (
{(provider || model) && (
<div className="space-y-1">
<p className="text-2xs uppercase tracking-wide text-muted-foreground">{t("form.model")}</p>
<Badge variant="outline">{model}</Badge>
<p className="text-2xs uppercase tracking-wide text-muted-foreground">
{provider && model
? `${t("form.provider")} / ${t("form.model")}`
: t(provider ? "form.provider" : "form.model")}
</p>
<div className="flex flex-wrap gap-2">
{provider && <Badge variant="outline">{provider}</Badge>}
{model && <Badge variant="outline">{model}</Badge>}
</div>
</div>
)}
</div>
@@ -0,0 +1,39 @@
import type { HookFormData } from "@/schemas/hooks.schema";
function parseHeaders(raw: string | undefined): Record<string, unknown> {
const trimmed = (raw ?? "").trim();
if (!trimmed) return {};
try {
const parsed = JSON.parse(trimmed);
if (parsed && typeof parsed === "object" && !Array.isArray(parsed)) {
return parsed as Record<string, unknown>;
}
throw new Error("headers must be a JSON object");
} catch (err) {
// The original parser message is included verbatim in the user-facing error.
// eslint-disable-next-line preserve-caught-error
throw new Error(
"Invalid headers JSON: " + (err instanceof Error ? err.message : String(err)),
);
}
}
export function buildHookConfig(data: HookFormData): Record<string, unknown> {
if (data.handler_type === "http") {
return {
url: data.url ?? "",
method: data.method ?? "POST",
headers: parseHeaders(data.headers),
body_template: data.body_template ?? "",
};
}
if (data.handler_type === "script") {
return { source: data.script_source ?? "" };
}
return {
prompt_template: data.prompt_template ?? "",
provider: data.provider ?? "",
model: data.model ?? "",
max_invocations_per_turn: data.max_invocations_per_turn ?? 5,
};
}
+3 -43
View File
@@ -23,6 +23,7 @@ import { HookTestPanel } from "./components/hook-test-panel";
import { HookOverviewTab } from "./components/hook-overview-tab";
import { HookHistoryTable } from "./components/hook-history-table";
import { BetaInfoCard } from "./components/beta-info-card";
import { buildHookConfig } from "./hook-form-config";
import type { HookFormData } from "@/schemas/hooks.schema";
const HOOK_EVENTS = [
@@ -30,47 +31,6 @@ const HOOK_EVENTS = [
"post_tool_use", "post_model_response", "stop", "subagent_start", "subagent_stop",
] as const;
// parseHeaders accepts an empty string, an empty object string, or a JSON
// object. Returns {} for empty/whitespace-only input. Throws a typed Error
// with a friendly message on malformed JSON so the caller can surface via toast.
function parseHeaders(raw: string | undefined): Record<string, unknown> {
const trimmed = (raw ?? "").trim();
if (!trimmed) return {};
try {
const parsed = JSON.parse(trimmed);
if (parsed && typeof parsed === "object" && !Array.isArray(parsed)) {
return parsed as Record<string, unknown>;
}
throw new Error("headers must be a JSON object");
} catch (err) {
// eslint-disable-next-line preserve-caught-error -- JSON.parse error message already captured verbatim in thrown message
throw new Error(
"Invalid headers JSON: " + (err instanceof Error ? err.message : String(err)),
);
}
}
function buildConfig(data: HookFormData): Record<string, unknown> {
if (data.handler_type === "http") {
return {
url: data.url ?? "",
method: data.method ?? "POST",
headers: parseHeaders(data.headers),
body_template: data.body_template ?? "",
};
}
if (data.handler_type === "script") {
// Backend goja handler reads cfg.Config.source (Phase 03). Zod caps at 32 KiB.
return { source: data.script_source ?? "" };
}
// prompt
return {
prompt_template: data.prompt_template ?? "",
model: data.model ?? "haiku",
max_invocations_per_turn: data.max_invocations_per_turn ?? 5,
};
}
export function HooksPage() {
// Route params — single source of truth (CLAUDE.md)
const { id: detailId } = useParams<{ id: string }>();
@@ -114,7 +74,7 @@ export function HooksPage() {
const handleCreate = async (data: HookFormData) => {
let config: Record<string, unknown>;
try {
config = buildConfig(data);
config = buildHookConfig(data);
} catch (err) {
toast.error(err instanceof Error ? err.message : String(err));
return;
@@ -139,7 +99,7 @@ export function HooksPage() {
if (!editTarget) return;
let config: Record<string, unknown>;
try {
config = buildConfig(data);
config = buildHookConfig(data);
} catch (err) {
toast.error(err instanceof Error ? err.message : String(err));
return;
+15
View File
@@ -38,6 +38,7 @@ export const hookFormSchema = z
headers: z.string().optional(), // JSON string
body_template: z.string().optional(),
prompt_template: z.string().optional(),
provider: z.string().optional(),
model: z.string().optional(),
max_invocations_per_turn: z.number().int().min(1).max(20).optional(),
// Script handler source (ES5.1 JavaScript). Cap mirrors backend 32 KiB
@@ -73,6 +74,20 @@ export const hookFormSchema = z
message: "validation.promptTemplateRequired",
});
}
if (!data.provider?.trim()) {
ctx.addIssue({
code: z.ZodIssueCode.custom,
path: ["provider"],
message: "validation.promptProviderRequired",
});
}
if (!data.model?.trim()) {
ctx.addIssue({
code: z.ZodIssueCode.custom,
path: ["model"],
message: "validation.promptModelRequired",
});
}
}
if (data.handler_type === "script" && !data.script_source?.trim()) {
ctx.addIssue({