mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-19 18:15:41 +00:00
merge(dev): resolve usage cap conflicts
This commit is contained in:
commit
7fee3073df
111 files changed
+3956
-248
No files matched your search
@@ -0,0 +1,131 @@
|
||||
package cmd
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/url"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/go-rod/rod/lib/proto"
|
||||
"github.com/google/uuid"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/store"
|
||||
"github.com/nextlevelbuilder/goclaw/pkg/browser"
|
||||
)
|
||||
|
||||
type storeBrowserCookieProvider struct {
|
||||
cookies store.BrowserCookieStore
|
||||
}
|
||||
|
||||
func newStoreBrowserCookieProvider(cookies store.BrowserCookieStore) browser.CookieProvider {
|
||||
if cookies == nil {
|
||||
return nil
|
||||
}
|
||||
return &storeBrowserCookieProvider{cookies: cookies}
|
||||
}
|
||||
|
||||
func (p *storeBrowserCookieProvider) CookiesForURL(ctx context.Context, scope browser.BrowserScope, targetURL string) ([]*proto.NetworkCookieParam, error) {
|
||||
u, err := url.Parse(targetURL)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("parse target url: %w", err)
|
||||
}
|
||||
if u.Scheme != "http" && u.Scheme != "https" {
|
||||
return nil, nil
|
||||
}
|
||||
storeScope, err := browserScopeToCookieScope(scope)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
cookies, err := p.cookies.List(ctx, storeScope, store.BrowserCookieFilter{})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
host := strings.ToLower(u.Hostname())
|
||||
path := u.EscapedPath()
|
||||
if path == "" {
|
||||
path = "/"
|
||||
}
|
||||
now := time.Now().UTC()
|
||||
params := make([]*proto.NetworkCookieParam, 0, len(cookies))
|
||||
for _, c := range cookies {
|
||||
if !browserCookieMatchesURL(c, host, path, now) {
|
||||
continue
|
||||
}
|
||||
param := &proto.NetworkCookieParam{
|
||||
Name: c.Name,
|
||||
Value: c.Value,
|
||||
URL: targetURL,
|
||||
Path: c.Path,
|
||||
Secure: c.Secure,
|
||||
HTTPOnly: c.HTTPOnly,
|
||||
SameSite: browserCookieSameSite(c.SameSite),
|
||||
}
|
||||
if strings.HasPrefix(c.Domain, ".") {
|
||||
param.Domain = c.Domain
|
||||
}
|
||||
if c.ExpiresAt != nil {
|
||||
param.Expires = proto.TimeSinceEpoch(float64(c.ExpiresAt.Unix()))
|
||||
}
|
||||
params = append(params, param)
|
||||
}
|
||||
return params, nil
|
||||
}
|
||||
|
||||
func browserScopeToCookieScope(scope browser.BrowserScope) (store.BrowserCookieScope, error) {
|
||||
tenantID := store.MasterTenantID
|
||||
if strings.TrimSpace(scope.TenantID) != "" {
|
||||
parsed, err := uuid.Parse(strings.TrimSpace(scope.TenantID))
|
||||
if err != nil {
|
||||
return store.BrowserCookieScope{}, fmt.Errorf("invalid browser tenant scope: %w", err)
|
||||
}
|
||||
tenantID = parsed
|
||||
}
|
||||
cookieScope := store.BrowserCookieScope{
|
||||
TenantID: tenantID,
|
||||
UserID: strings.TrimSpace(scope.UserID),
|
||||
AgentID: strings.TrimSpace(scope.AgentID),
|
||||
}
|
||||
if err := cookieScope.Validate(); err != nil {
|
||||
return store.BrowserCookieScope{}, err
|
||||
}
|
||||
return cookieScope, nil
|
||||
}
|
||||
|
||||
func browserCookieMatchesURL(c store.BrowserCookie, host, requestPath string, now time.Time) bool {
|
||||
domain := strings.ToLower(strings.TrimSpace(c.Domain))
|
||||
if domain == "" {
|
||||
return false
|
||||
}
|
||||
if c.ExpiresAt != nil && !c.ExpiresAt.After(now) {
|
||||
return false
|
||||
}
|
||||
hostOnly := !strings.HasPrefix(domain, ".")
|
||||
matchDomain := strings.TrimPrefix(domain, ".")
|
||||
if hostOnly {
|
||||
if host != matchDomain {
|
||||
return false
|
||||
}
|
||||
} else if host != matchDomain && !strings.HasSuffix(host, "."+matchDomain) {
|
||||
return false
|
||||
}
|
||||
cookiePath := c.Path
|
||||
if cookiePath == "" {
|
||||
cookiePath = "/"
|
||||
}
|
||||
return requestPath == cookiePath || strings.HasPrefix(requestPath, strings.TrimRight(cookiePath, "/")+"/")
|
||||
}
|
||||
|
||||
func browserCookieSameSite(value string) proto.NetworkCookieSameSite {
|
||||
switch strings.ToLower(strings.TrimSpace(value)) {
|
||||
case "strict":
|
||||
return proto.NetworkCookieSameSiteStrict
|
||||
case "lax":
|
||||
return proto.NetworkCookieSameSiteLax
|
||||
case "none", "no_restriction", "no-restriction":
|
||||
return proto.NetworkCookieSameSiteNone
|
||||
default:
|
||||
return ""
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,76 @@
|
||||
package cmd
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/store"
|
||||
"github.com/nextlevelbuilder/goclaw/pkg/browser"
|
||||
)
|
||||
|
||||
type fakeStoreCookieProviderStore struct {
|
||||
items []store.BrowserCookie
|
||||
scope store.BrowserCookieScope
|
||||
}
|
||||
|
||||
func (f *fakeStoreCookieProviderStore) Upsert(context.Context, store.BrowserCookieScope, []store.BrowserCookie) (int, error) {
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
func (f *fakeStoreCookieProviderStore) List(_ context.Context, scope store.BrowserCookieScope, _ store.BrowserCookieFilter) ([]store.BrowserCookie, error) {
|
||||
f.scope = scope
|
||||
return f.items, nil
|
||||
}
|
||||
|
||||
func (f *fakeStoreCookieProviderStore) Delete(context.Context, store.BrowserCookieScope, store.BrowserCookieFilter) (int, error) {
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
func TestStoreBrowserCookieProviderFiltersAndConvertsCookies(t *testing.T) {
|
||||
tenantID := uuid.New()
|
||||
expiresAt := time.Now().UTC().Add(time.Hour)
|
||||
fake := &fakeStoreCookieProviderStore{items: []store.BrowserCookie{
|
||||
{Domain: ".example.com", Name: "parent", Path: "/", Value: "parent-secret", Secure: true, HTTPOnly: true, SameSite: "Lax", ExpiresAt: &expiresAt},
|
||||
{Domain: "app.example.com", Name: "host", Path: "/app", Value: "host-secret"},
|
||||
{Domain: "other.example.com", Name: "skip", Path: "/", Value: "skip"},
|
||||
}}
|
||||
provider := newStoreBrowserCookieProvider(fake)
|
||||
|
||||
got, err := provider.CookiesForURL(context.Background(), browser.BrowserScope{
|
||||
TenantID: tenantID.String(),
|
||||
UserID: "user-a",
|
||||
AgentID: "agent-a",
|
||||
}, "https://app.example.com/app/page")
|
||||
if err != nil {
|
||||
t.Fatalf("CookiesForURL: %v", err)
|
||||
}
|
||||
if fake.scope.TenantID != tenantID || fake.scope.UserID != "user-a" || fake.scope.AgentID != "agent-a" {
|
||||
t.Fatalf("scope = %+v", fake.scope)
|
||||
}
|
||||
if len(got) != 2 {
|
||||
t.Fatalf("cookies len = %d, want 2: %+v", len(got), got)
|
||||
}
|
||||
if got[0].Name != "parent" || got[0].Domain != ".example.com" || !got[0].HTTPOnly {
|
||||
t.Fatalf("parent cookie mismatch: %+v", got[0])
|
||||
}
|
||||
if got[1].Name != "host" || got[1].Domain != "" || got[1].URL == "" {
|
||||
t.Fatalf("host cookie should be URL-scoped: %+v", got[1])
|
||||
}
|
||||
}
|
||||
|
||||
func TestBrowserCookieMatchesURLRejectsExpiredAndWrongPath(t *testing.T) {
|
||||
now := time.Now().UTC()
|
||||
expiredAt := now.Add(-time.Minute)
|
||||
if browserCookieMatchesURL(store.BrowserCookie{Domain: "example.com", Name: "expired", Path: "/", ExpiresAt: &expiredAt}, "example.com", "/", now) {
|
||||
t.Fatal("expired cookie matched")
|
||||
}
|
||||
if browserCookieMatchesURL(store.BrowserCookie{Domain: "example.com", Name: "wrong-path", Path: "/admin"}, "example.com", "/app", now) {
|
||||
t.Fatal("wrong-path cookie matched")
|
||||
}
|
||||
if !browserCookieMatchesURL(store.BrowserCookie{Domain: ".example.com", Name: "sub", Path: "/"}, "app.example.com", "/app", now) {
|
||||
t.Fatal("parent-domain cookie did not match subdomain")
|
||||
}
|
||||
}
|
||||
@@ -173,6 +173,9 @@ func runGateway() {
|
||||
}
|
||||
|
||||
pgStores, traceCollector, snapshotWorker := setupStoresAndTracing(cfg, dataDir, msgBus)
|
||||
if browserMgr != nil && pgStores != nil && pgStores.BrowserCookies != nil && cfg.Tools.Browser.CookieSyncEnabled {
|
||||
browserMgr.SetCookieProvider(newStoreBrowserCookieProvider(pgStores.BrowserCookies))
|
||||
}
|
||||
|
||||
// Recover from crashes: flip ghost 'summoning' rows to 'summon_failed'.
|
||||
// Summon goroutines don't survive process restart; stale DB rows would trap the UI.
|
||||
|
||||
@@ -84,19 +84,17 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents
|
||||
|
||||
// Inbound debounce: merge rapid messages from the same sender before processing.
|
||||
// Matching TS createInboundDebouncer from src/auto-reply/inbound-debounce.ts.
|
||||
debounceMs := cfg.Gateway.InboundDebounceMs
|
||||
if debounceMs == 0 {
|
||||
debounceMs = 1000 // default: 1000ms
|
||||
}
|
||||
debouncer := bus.NewInboundDebouncer(
|
||||
time.Duration(debounceMs)*time.Millisecond,
|
||||
debouncer := bus.NewInboundDebouncerFunc(
|
||||
func(msg bus.InboundMessage) time.Duration {
|
||||
return resolveInboundDebounceDelay(ctx, msg, deps)
|
||||
},
|
||||
func(msg bus.InboundMessage) {
|
||||
processNormalMessage(ctx, msg, deps)
|
||||
},
|
||||
)
|
||||
defer debouncer.Stop()
|
||||
|
||||
slog.Info("inbound debounce configured", "debounce_ms", debounceMs)
|
||||
slog.Info("inbound debounce configured", "global_debounce_ms", cfg.Gateway.InboundDebounceMs, "agent_override", true)
|
||||
|
||||
// Track background goroutines (subagent announces, teammate messages)
|
||||
// so shutdown can wait for in-flight work to complete.
|
||||
@@ -141,6 +139,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents
|
||||
}
|
||||
|
||||
// --- Normal messages: route through debouncer ---
|
||||
prepareInboundDebounceMessage(&msg, deps)
|
||||
debouncer.Push(msg)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,62 @@
|
||||
package cmd
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/bus"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/store"
|
||||
)
|
||||
|
||||
func prepareInboundDebounceMessage(msg *bus.InboundMessage, deps *ConsumerDeps) {
|
||||
if msg == nil || deps == nil || deps.Cfg == nil || msg.AgentID != "" {
|
||||
return
|
||||
}
|
||||
msg.AgentID = resolveAgentRoute(deps.Cfg, msg.Channel, msg.ChatID, msg.PeerKind)
|
||||
}
|
||||
|
||||
func resolveInboundDebounceDelay(ctx context.Context, msg bus.InboundMessage, deps *ConsumerDeps) time.Duration {
|
||||
debounceMs := 0
|
||||
if deps != nil && deps.Cfg != nil {
|
||||
debounceMs = deps.Cfg.Gateway.InboundDebounceMs
|
||||
}
|
||||
if deps == nil || deps.AgentStore == nil || msg.AgentID == "" {
|
||||
return inboundDebounceDuration(debounceMs)
|
||||
}
|
||||
|
||||
agentCtx := ctx
|
||||
if msg.TenantID != uuid.Nil {
|
||||
agentCtx = store.WithTenantID(agentCtx, msg.TenantID)
|
||||
} else {
|
||||
agentCtx = store.WithTenantID(agentCtx, store.MasterTenantID)
|
||||
}
|
||||
|
||||
agentData, err := getInboundDebounceAgent(agentCtx, deps.AgentStore, msg.AgentID)
|
||||
if err != nil || agentData == nil {
|
||||
if err != nil {
|
||||
slog.Debug("inbound debounce: agent config unavailable", "agent", msg.AgentID, "error", err)
|
||||
}
|
||||
return inboundDebounceDuration(debounceMs)
|
||||
}
|
||||
if overrideMs, ok := agentData.ParseInboundDebounceMs(); ok {
|
||||
debounceMs = overrideMs
|
||||
}
|
||||
return inboundDebounceDuration(debounceMs)
|
||||
}
|
||||
|
||||
func getInboundDebounceAgent(ctx context.Context, agentStore store.AgentStore, agentID string) (*store.AgentData, error) {
|
||||
if parsed, err := uuid.Parse(agentID); err == nil && parsed != uuid.Nil {
|
||||
return agentStore.GetByID(ctx, parsed)
|
||||
}
|
||||
return agentStore.GetByKey(ctx, agentID)
|
||||
}
|
||||
|
||||
func inboundDebounceDuration(ms int) time.Duration {
|
||||
if ms <= 0 {
|
||||
return 0
|
||||
}
|
||||
return time.Duration(ms) * time.Millisecond
|
||||
}
|
||||
@@ -84,6 +84,9 @@ func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg
|
||||
cronCtx, cancelCron := context.WithTimeout(context.Background(), jobTimeout)
|
||||
defer cancelCron()
|
||||
cronCtx = store.WithTenantID(cronCtx, job.TenantID)
|
||||
if job.Payload.CredentialUserID != "" {
|
||||
cronCtx = store.WithCredentialUserID(cronCtx, job.Payload.CredentialUserID)
|
||||
}
|
||||
|
||||
// Reset session before each cron run to prevent tool errors from previous
|
||||
// runs from polluting the context and blocking future executions (#294).
|
||||
|
||||
@@ -0,0 +1,66 @@
|
||||
package cmd
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/google/uuid"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/agent"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/config"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/scheduler"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/store"
|
||||
)
|
||||
|
||||
func TestCronJobHandlerInjectsPayloadCredentialUserID(t *testing.T) {
|
||||
wantCredentialUserID := "tenant-user-123"
|
||||
var gotCredentialUserID string
|
||||
|
||||
sched := scheduler.NewScheduler(
|
||||
scheduler.DefaultLanes(),
|
||||
scheduler.QueueConfig{
|
||||
Mode: scheduler.QueueModeQueue,
|
||||
Cap: 1,
|
||||
Drop: scheduler.DropOld,
|
||||
DebounceMs: 0,
|
||||
MaxConcurrent: 1,
|
||||
},
|
||||
func(ctx context.Context, req agent.RunRequest) (*agent.RunResult, error) {
|
||||
gotCredentialUserID = store.CredentialUserIDFromContext(ctx)
|
||||
return &agent.RunResult{Content: "ok"}, nil
|
||||
},
|
||||
)
|
||||
defer sched.Stop()
|
||||
|
||||
handler := makeCronJobHandler(
|
||||
sched,
|
||||
nil,
|
||||
&config.Config{},
|
||||
nil,
|
||||
nil,
|
||||
nil,
|
||||
)
|
||||
|
||||
result, err := handler(&store.CronJob{
|
||||
ID: uuid.NewString(),
|
||||
TenantID: uuid.New(),
|
||||
Name: "credentialed-report",
|
||||
AgentID: "reporter",
|
||||
UserID: "group:telegram:-100123",
|
||||
Stateless: true,
|
||||
Payload: store.CronPayload{
|
||||
Kind: "agent_turn",
|
||||
Message: "run gh issue list",
|
||||
CredentialUserID: wantCredentialUserID,
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("cron handler returned error: %v", err)
|
||||
}
|
||||
if result == nil || result.Content != "ok" {
|
||||
t.Fatalf("cron result = %#v, want content ok", result)
|
||||
}
|
||||
if gotCredentialUserID != wantCredentialUserID {
|
||||
t.Fatalf("credential user ID in scheduled context = %q, want %q", gotCredentialUserID, wantCredentialUserID)
|
||||
}
|
||||
}
|
||||
@@ -96,6 +96,9 @@ func (d *gatewayDeps) wireHTTPHandlersOnServer(
|
||||
if h.secureCLIGrant != nil {
|
||||
d.server.SetSecureCLIGrantHandler(h.secureCLIGrant)
|
||||
}
|
||||
if d.pgStores != nil && d.pgStores.BrowserCookies != nil {
|
||||
d.server.SetBrowserCookiesHandler(httpapi.NewBrowserCookiesHandler(d.pgStores.BrowserCookies))
|
||||
}
|
||||
|
||||
// Activity audit log API
|
||||
if d.pgStores.Activity != nil {
|
||||
|
||||
@@ -78,6 +78,9 @@ func seedConfigForContext(ctx context.Context, sc store.SystemConfigStore, cfg *
|
||||
set(key, fmt.Sprintf("%d", val))
|
||||
}
|
||||
}
|
||||
setIntAllowZero := func(key string, val int) {
|
||||
set(key, fmt.Sprintf("%d", val))
|
||||
}
|
||||
setBool := func(key string, val *bool) {
|
||||
if val != nil {
|
||||
set(key, fmt.Sprintf("%t", *val))
|
||||
@@ -102,7 +105,7 @@ func seedConfigForContext(ctx context.Context, sc store.SystemConfigStore, cfg *
|
||||
setInt("gateway.rate_limit_rpm", cfg.Gateway.RateLimitRPM)
|
||||
setInt("gateway.max_message_chars", cfg.Gateway.MaxMessageChars)
|
||||
set("gateway.injection_action", cfg.Gateway.InjectionAction)
|
||||
setInt("gateway.inbound_debounce_ms", cfg.Gateway.InboundDebounceMs)
|
||||
setIntAllowZero("gateway.inbound_debounce_ms", cfg.Gateway.InboundDebounceMs)
|
||||
setBool("gateway.block_reply", cfg.Gateway.BlockReply)
|
||||
setBool("gateway.tool_status", cfg.Gateway.ToolStatus)
|
||||
setInt("gateway.task_recovery_interval_sec", cfg.Gateway.TaskRecoveryIntervalSec)
|
||||
@@ -115,6 +118,13 @@ func seedConfigForContext(ctx context.Context, sc store.SystemConfigStore, cfg *
|
||||
set("tools.profile", cfg.Tools.Profile)
|
||||
setInt("tools.rate_limit_per_hour", cfg.Tools.RateLimitPerHour)
|
||||
setBool("tools.scrub_credentials", cfg.Tools.ScrubCredentials)
|
||||
set("tools.browser.enabled", fmt.Sprintf("%t", cfg.Tools.Browser.Enabled))
|
||||
set("tools.browser.headless", fmt.Sprintf("%t", cfg.Tools.Browser.Headless))
|
||||
set("tools.browser.remote_url", cfg.Tools.Browser.RemoteURL)
|
||||
setInt("tools.browser.action_timeout_ms", cfg.Tools.Browser.ActionTimeoutMs)
|
||||
setIntAllowZero("tools.browser.idle_timeout_ms", cfg.Tools.Browser.IdleTimeoutMs)
|
||||
setInt("tools.browser.max_pages", cfg.Tools.Browser.MaxPages)
|
||||
set("tools.browser.cookie_sync_enabled", fmt.Sprintf("%t", cfg.Tools.Browser.CookieSyncEnabled))
|
||||
|
||||
// TTS
|
||||
set("tts.provider", cfg.Tts.Provider)
|
||||
|
||||
@@ -0,0 +1,48 @@
|
||||
package cmd
|
||||
|
||||
import (
|
||||
"context"
|
||||
"maps"
|
||||
"testing"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/config"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/store"
|
||||
)
|
||||
|
||||
type captureSystemConfigStore struct {
|
||||
data map[string]string
|
||||
}
|
||||
|
||||
func (s *captureSystemConfigStore) Get(_ context.Context, key string) (string, error) {
|
||||
return s.data[key], nil
|
||||
}
|
||||
|
||||
func (s *captureSystemConfigStore) Set(_ context.Context, key, value string) error {
|
||||
s.data[key] = value
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *captureSystemConfigStore) Delete(_ context.Context, key string) error {
|
||||
delete(s.data, key)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *captureSystemConfigStore) List(_ context.Context) (map[string]string, error) {
|
||||
out := make(map[string]string, len(s.data))
|
||||
maps.Copy(out, s.data)
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func TestSeedConfigForContextPersistsZeroInboundDebounce(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
sc := &captureSystemConfigStore{data: map[string]string{}}
|
||||
cfg := config.Default()
|
||||
cfg.Gateway.InboundDebounceMs = 0
|
||||
|
||||
seedConfigForContext(store.WithTenantID(context.Background(), store.MasterTenantID), sc, cfg, false)
|
||||
|
||||
if got := sc.data["gateway.inbound_debounce_ms"]; got != "0" {
|
||||
t.Fatalf("gateway.inbound_debounce_ms = %q, want 0", got)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user