From 90b7396d74f55c43e333871bd0c07f0ec26ef15a Mon Sep 17 00:00:00 2001 From: viettranx Date: Sat, 28 Mar 2026 18:15:46 +0700 Subject: [PATCH] fix(bus,cron): panic recovery + deadlock fix + unit tests for 9 packages MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bug fixes: - bus: add panic recovery in Broadcast() — panicking subscriber no longer crashes entire event bus goroutine - cron: fix deadlock in RunJob() — recordRun called while holding mutex, extracted recordRunLocked for callers already holding lock - cron: add run log recording to executeJobByID — automatic scheduler was not populating run log, only manual RunJob did New tests (~170 cases across 9 packages): - crypto: roundtrip, key derivation (hex/b64/raw), nonce uniqueness, backward compat, wrong key, corruption - permissions: role hierarchy, RoleFromScopes, CanAccess, scope precedence - bus: pub/sub delivery, buffer full, panic recovery, concurrent safety - providers/retry: IsRetryableError, backoff, jitter, Retry-After, context cancellation, hook callback - sessions: idempotency, concurrent writes, defensive copy, save/load roundtrip, metadata accumulation - scheduler: draining, drop policies, adaptive throttle, stale completion, lane concurrency, debounce, interrupt mode - config: JSON5 parsing, env overrides, FlexibleStringSlice, owner IDs - cron: schedule validation, computeNextRun, CRUD, job execution, failure tracking, persistence roundtrip - agent: history sanitization edge cases — all-tool history, partial results, cross-turn dedup, orphaned tools, 1000-msg performance --- internal/agent/loop_history_edge_test.go | 264 ++++++++++++ internal/bus/bus.go | 19 +- internal/bus/bus_test.go | 263 ++++++++++++ internal/config/config_load_test.go | 201 +++++++++ internal/cron/service_execution.go | 11 +- internal/cron/service_test.go | 357 ++++++++++++++++ internal/crypto/aes_test.go | 261 ++++++++++++ internal/permissions/policy_test.go | 280 ++++++++++++ internal/providers/retry_test.go | 330 +++++++++++++++ internal/scheduler/scheduler_queue_test.go | 466 ++++++++++++++++++++ internal/sessions/manager_test.go | 467 +++++++++++++++++++++ 11 files changed, 2915 insertions(+), 4 deletions(-) create mode 100644 internal/agent/loop_history_edge_test.go create mode 100644 internal/bus/bus_test.go create mode 100644 internal/config/config_load_test.go create mode 100644 internal/cron/service_test.go create mode 100644 internal/crypto/aes_test.go create mode 100644 internal/permissions/policy_test.go create mode 100644 internal/providers/retry_test.go create mode 100644 internal/scheduler/scheduler_queue_test.go create mode 100644 internal/sessions/manager_test.go diff --git a/internal/agent/loop_history_edge_test.go b/internal/agent/loop_history_edge_test.go new file mode 100644 index 00000000..95faf606 --- /dev/null +++ b/internal/agent/loop_history_edge_test.go @@ -0,0 +1,264 @@ +package agent + +import ( + "strings" + "testing" + + "github.com/nextlevelbuilder/goclaw/internal/providers" +) + +// --- sanitizeHistory: all-tool-message history --- +// After aggressive truncation, history could be ALL tool messages. +// Should return nil (no useful messages), not crash. + +func TestSanitizeHistory_AllToolMessages(t *testing.T) { + msgs := []providers.Message{ + {Role: "tool", Content: "result1", ToolCallID: "tc1"}, + {Role: "tool", Content: "result2", ToolCallID: "tc2"}, + {Role: "tool", Content: "result3", ToolCallID: "tc3"}, + } + + result, dropped := sanitizeHistory(msgs) + if result != nil { + t.Fatalf("all-tool history should return nil, got %d messages", len(result)) + } + if dropped != 3 { + t.Fatalf("expected 3 dropped, got %d", dropped) + } +} + +// --- sanitizeHistory: single user message (no tools) --- + +func TestSanitizeHistory_SingleUserMessage(t *testing.T) { + msgs := []providers.Message{ + {Role: "user", Content: "hello"}, + } + + result, dropped := sanitizeHistory(msgs) + if len(result) != 1 { + t.Fatalf("expected 1 message, got %d", len(result)) + } + if dropped != 0 { + t.Fatalf("expected 0 dropped, got %d", dropped) + } +} + +// --- sanitizeHistory: assistant with tool_calls but ALL results missing --- +// Every tool_call should get a synthesized "[Tool result missing]" placeholder. + +func TestSanitizeHistory_AllToolResultsMissing(t *testing.T) { + msgs := []providers.Message{ + {Role: "user", Content: "do something"}, + {Role: "assistant", Content: "", ToolCalls: []providers.ToolCall{ + {ID: "tc1", Name: "read_file", Arguments: map[string]any{"path": "a.go"}}, + {ID: "tc2", Name: "read_file", Arguments: map[string]any{"path": "b.go"}}, + {ID: "tc3", Name: "read_file", Arguments: map[string]any{"path": "c.go"}}, + }}, + // No tool results at all — all missing + {Role: "user", Content: "next message"}, + } + + result, dropped := sanitizeHistory(msgs) + + // Should have: user + assistant + 3 synthesized tool results + user + if len(result) != 6 { + t.Fatalf("expected 6 messages (user + assistant + 3 synth + user), got %d", len(result)) + } + if dropped != 3 { + t.Fatalf("expected 3 synthesized (counted as dropped), got %d", dropped) + } + + // Verify synthesized messages + for i := 2; i <= 4; i++ { + if result[i].Role != "tool" { + t.Fatalf("message %d: expected role 'tool', got %q", i, result[i].Role) + } + if !strings.Contains(result[i].Content, "missing") { + t.Fatalf("message %d: expected 'missing' placeholder, got %q", i, result[i].Content) + } + } +} + +// --- sanitizeHistory: interleaved tool results from wrong assistant --- +// tool_result IDs that don't match the preceding assistant's tool_calls +// should be dropped. + +func TestSanitizeHistory_ToolResultsFromWrongAssistant(t *testing.T) { + msgs := []providers.Message{ + {Role: "user", Content: "first"}, + {Role: "assistant", Content: "", ToolCalls: []providers.ToolCall{ + {ID: "tc_a", Name: "read_file", Arguments: map[string]any{}}, + }}, + // Tool result with wrong ID (from a different assistant turn) + {Role: "tool", Content: "wrong result", ToolCallID: "tc_z"}, + // Correct result + // (missing — will be synthesized) + {Role: "user", Content: "next"}, + } + + result, dropped := sanitizeHistory(msgs) + + // tc_z dropped, tc_a synthesized + if dropped != 2 { + t.Fatalf("expected 2 dropped (1 mismatched + 1 synthesized), got %d", dropped) + } + + // Should have: user + assistant + synth(tc_a) + user + if len(result) != 4 { + t.Fatalf("expected 4 messages, got %d", len(result)) + } + if result[2].ToolCallID != "tc_a" { + t.Fatalf("synthesized result should have tc_a, got %q", result[2].ToolCallID) + } +} + +// --- sanitizeHistory: multiple tool_calls with partial results --- +// 3 tool_calls but only 1 result provided → 2 synthesized + +func TestSanitizeHistory_PartialToolResults(t *testing.T) { + msgs := []providers.Message{ + {Role: "user", Content: "go"}, + {Role: "assistant", Content: "", ToolCalls: []providers.ToolCall{ + {ID: "tc1", Name: "read_file", Arguments: map[string]any{"path": "a"}}, + {ID: "tc2", Name: "write_file", Arguments: map[string]any{"path": "b"}}, + {ID: "tc3", Name: "exec", Arguments: map[string]any{"cmd": "ls"}}, + }}, + {Role: "tool", Content: "content of a", ToolCallID: "tc1"}, + // tc2 and tc3 missing + {Role: "user", Content: "done"}, + } + + result, dropped := sanitizeHistory(msgs) + + // user + assistant + tc1_result + tc2_synth + tc3_synth + user + if len(result) != 6 { + t.Fatalf("expected 6 messages, got %d", len(result)) + } + if dropped != 2 { + t.Fatalf("expected 2 synthesized, got %d", dropped) + } + // Verify order: real result first, then synthesized + if result[2].ToolCallID != "tc1" || strings.Contains(result[2].Content, "missing") { + t.Fatal("tc1 should be the real result, not synthesized") + } + if result[3].ToolCallID != "tc2" || !strings.Contains(result[3].Content, "missing") { + t.Fatal("tc2 should be synthesized") + } + if result[4].ToolCallID != "tc3" || !strings.Contains(result[4].Content, "missing") { + t.Fatal("tc3 should be synthesized") + } +} + +// --- sanitizeHistory: tool message between two user messages (orphaned mid-history) --- + +func TestSanitizeHistory_OrphanedToolBetweenUsers(t *testing.T) { + msgs := []providers.Message{ + {Role: "user", Content: "first"}, + {Role: "assistant", Content: "response"}, + {Role: "tool", Content: "orphaned", ToolCallID: "tc_orphan"}, // no preceding tool_calls + {Role: "user", Content: "second"}, + } + + result, dropped := sanitizeHistory(msgs) + + if dropped != 1 { + t.Fatalf("expected 1 dropped orphan, got %d", dropped) + } + // user + assistant + user (orphan dropped) + if len(result) != 3 { + t.Fatalf("expected 3 messages, got %d", len(result)) + } +} + +// --- sanitizeHistory: dedup with 2 identical IDs across 2 turns --- +// Turn 1: ID "dup" → kept as-is (first occurrence) +// Turn 2: ID "dup" → rewritten to "dup_dedup_0", tool result matched via idQueue + +func TestSanitizeHistory_DedupAcrossTwoTurns(t *testing.T) { + msgs := []providers.Message{ + // Turn 1 + {Role: "user", Content: "1"}, + {Role: "assistant", ToolCalls: []providers.ToolCall{{ID: "dup", Name: "read_file", Arguments: map[string]any{}}}}, + {Role: "tool", Content: "r1", ToolCallID: "dup"}, + // Turn 2 — same ID + {Role: "user", Content: "2"}, + {Role: "assistant", ToolCalls: []providers.ToolCall{{ID: "dup", Name: "read_file", Arguments: map[string]any{}}}}, + {Role: "tool", Content: "r2", ToolCallID: "dup"}, + } + + result, dropped := sanitizeHistory(msgs) + + // All 6 messages should be preserved (dedup rewrites turn 2's ID) + if len(result) != 6 { + t.Fatalf("expected 6 messages preserved via dedup, got %d", len(result)) + } + if dropped != 0 { + t.Fatalf("expected 0 dropped, got %d", dropped) + } + + // Turn 1 assistant should keep original ID + if result[1].ToolCalls[0].ID != "dup" { + t.Fatalf("turn 1 ID should be 'dup', got %q", result[1].ToolCalls[0].ID) + } + // Turn 2 assistant should have rewritten ID + if result[4].ToolCalls[0].ID == "dup" { + t.Fatal("turn 2 ID should be rewritten, still 'dup'") + } + // Turn 2 tool result should match the rewritten ID + if result[5].ToolCallID != result[4].ToolCalls[0].ID { + t.Fatalf("turn 2 tool result ID %q should match assistant ID %q", + result[5].ToolCallID, result[4].ToolCalls[0].ID) + } +} + +// --- sanitizeHistory: performance with large history --- +// Verify sanitization doesn't degrade catastrophically with many messages. + +func TestSanitizeHistory_LargeHistory_Performance(t *testing.T) { + // Build 1000-message history with proper tool pairing + msgs := make([]providers.Message, 0, 1000) + for i := 0; i < 250; i++ { + tcID := "tc_" + strings.Repeat("x", 5) + "_" + string(rune('a'+i%26)) + string(rune('0'+i%10)) + msgs = append(msgs, + providers.Message{Role: "user", Content: "question " + tcID}, + providers.Message{Role: "assistant", ToolCalls: []providers.ToolCall{ + {ID: tcID, Name: "read_file", Arguments: map[string]any{"path": "file.go"}}, + }}, + providers.Message{Role: "tool", Content: "result", ToolCallID: tcID}, + providers.Message{Role: "assistant", Content: "answer"}, + ) + } + + result, dropped := sanitizeHistory(msgs) + + if dropped != 0 { + t.Fatalf("well-formed history should have 0 drops, got %d", dropped) + } + if len(result) != 1000 { + t.Fatalf("expected 1000 messages, got %d", len(result)) + } +} + +// --- sanitizeHistory: empty ToolCallID in tool result --- + +func TestSanitizeHistory_EmptyToolCallID(t *testing.T) { + msgs := []providers.Message{ + {Role: "user", Content: "go"}, + {Role: "assistant", ToolCalls: []providers.ToolCall{ + {ID: "tc1", Name: "read_file", Arguments: map[string]any{}}, + }}, + {Role: "tool", Content: "result", ToolCallID: ""}, // empty ID — won't match + } + + result, dropped := sanitizeHistory(msgs) + + // Empty ID tool result should be dropped (mismatched) + // tc1 should get a synthesized result + if dropped != 2 { + t.Fatalf("expected 2 dropped (1 mismatched + 1 synthesized), got %d", dropped) + } + // user + assistant + synth(tc1) + if len(result) != 3 { + t.Fatalf("expected 3 messages, got %d", len(result)) + } +} diff --git a/internal/bus/bus.go b/internal/bus/bus.go index a14a59b7..66a42a3e 100644 --- a/internal/bus/bus.go +++ b/internal/bus/bus.go @@ -2,6 +2,8 @@ package bus import ( "context" + "fmt" + "log/slog" "sync" ) @@ -113,11 +115,24 @@ func (mb *MessageBus) Unsubscribe(id string) { } // Broadcast sends an event to all subscribers (non-blocking per subscriber). +// Panicking handlers are caught and logged to prevent one bad subscriber +// from crashing the entire event bus. func (mb *MessageBus) Broadcast(event Event) { mb.subMu.RLock() defer mb.subMu.RUnlock() - for _, handler := range mb.subscribers { - handler(event) // handlers should be non-blocking + for id, handler := range mb.subscribers { + func() { + defer func() { + if r := recover(); r != nil { + slog.Error("bus: subscriber panicked", + "subscriber", id, + "event", event.Name, + "panic", fmt.Sprint(r), + ) + } + }() + handler(event) + }() } } diff --git a/internal/bus/bus_test.go b/internal/bus/bus_test.go new file mode 100644 index 00000000..dce2b81c --- /dev/null +++ b/internal/bus/bus_test.go @@ -0,0 +1,263 @@ +package bus + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" +) + +// --- Pub/Sub delivery --- + +func TestPublishInbound_ConsumeInbound(t *testing.T) { + mb := New() + defer mb.Close() + + msg := InboundMessage{Channel: "telegram", Content: "hello"} + mb.PublishInbound(msg) + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + got, ok := mb.ConsumeInbound(ctx) + if !ok { + t.Fatal("expected to consume message") + } + if got.Content != "hello" { + t.Fatalf("content mismatch: got %q, want %q", got.Content, "hello") + } +} + +func TestPublishOutbound_SubscribeOutbound(t *testing.T) { + mb := New() + defer mb.Close() + + msg := OutboundMessage{Channel: "discord", Content: "world"} + mb.PublishOutbound(msg) + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + got, ok := mb.SubscribeOutbound(ctx) + if !ok { + t.Fatal("expected to receive message") + } + if got.Content != "world" { + t.Fatalf("content mismatch: got %q, want %q", got.Content, "world") + } +} + +// --- Context cancellation --- + +func TestConsumeInbound_ContextCancelled(t *testing.T) { + mb := New() + defer mb.Close() + + ctx, cancel := context.WithCancel(context.Background()) + cancel() // cancel immediately + + _, ok := mb.ConsumeInbound(ctx) + if ok { + t.Fatal("expected false on cancelled context") + } +} + +// --- TryPublish non-blocking --- + +func TestTryPublishInbound_BufferFull(t *testing.T) { + mb := &MessageBus{ + inbound: make(chan InboundMessage, 1), // tiny buffer + outbound: make(chan OutboundMessage, 1), + handlers: make(map[string]MessageHandler), + subscribers: make(map[string]EventHandler), + } + + // First message fits + if !mb.TryPublishInbound(InboundMessage{Content: "1"}) { + t.Fatal("first message should succeed") + } + // Second message should be dropped (buffer full) + if mb.TryPublishInbound(InboundMessage{Content: "2"}) { + t.Fatal("second message should fail (buffer full)") + } +} + +func TestTryPublishOutbound_BufferFull(t *testing.T) { + mb := &MessageBus{ + inbound: make(chan InboundMessage, 1), + outbound: make(chan OutboundMessage, 1), + handlers: make(map[string]MessageHandler), + subscribers: make(map[string]EventHandler), + } + + if !mb.TryPublishOutbound(OutboundMessage{Content: "1"}) { + t.Fatal("first message should succeed") + } + if mb.TryPublishOutbound(OutboundMessage{Content: "2"}) { + t.Fatal("second message should fail (buffer full)") + } +} + +// --- Broadcast delivery --- + +func TestBroadcast_DeliveredToAllSubscribers(t *testing.T) { + mb := New() + defer mb.Close() + + var count atomic.Int32 + mb.Subscribe("sub1", func(e Event) { count.Add(1) }) + mb.Subscribe("sub2", func(e Event) { count.Add(1) }) + mb.Subscribe("sub3", func(e Event) { count.Add(1) }) + + mb.Broadcast(Event{Name: "test"}) + + if got := count.Load(); got != 3 { + t.Fatalf("expected 3 deliveries, got %d", got) + } +} + +// --- Broadcast panic recovery: panicking handler must NOT crash other subscribers --- + +func TestBroadcast_PanickingHandler_DoesNotCrashBus(t *testing.T) { + mb := New() + defer mb.Close() + + var delivered atomic.Int32 + + mb.Subscribe("panicker", func(e Event) { + panic("subscriber exploded") + }) + mb.Subscribe("normal", func(e Event) { + delivered.Add(1) + }) + + // This must not panic — the bus should catch the panicking handler + mb.Broadcast(Event{Name: "test"}) + + // The normal handler may or may not be called depending on iteration order, + // but the important thing is we didn't crash. + // Broadcast a second time to verify bus is still functional. + mb.Broadcast(Event{Name: "test2"}) + + // After two broadcasts, normal handler should have been called at least once + if got := delivered.Load(); got == 0 { + t.Fatal("normal handler should have been called at least once after two broadcasts") + } +} + +// --- Subscribe / Unsubscribe --- + +func TestUnsubscribe_StopsDelivery(t *testing.T) { + mb := New() + defer mb.Close() + + var count atomic.Int32 + mb.Subscribe("temp", func(e Event) { count.Add(1) }) + + mb.Broadcast(Event{Name: "before"}) + if count.Load() != 1 { + t.Fatal("expected delivery before unsubscribe") + } + + mb.Unsubscribe("temp") + mb.Broadcast(Event{Name: "after"}) + if count.Load() != 1 { + t.Fatal("expected no delivery after unsubscribe") + } +} + +func TestSubscribe_OverwritesPrevious(t *testing.T) { + mb := New() + defer mb.Close() + + var first, second atomic.Int32 + mb.Subscribe("id1", func(e Event) { first.Add(1) }) + mb.Subscribe("id1", func(e Event) { second.Add(1) }) // overwrite + + mb.Broadcast(Event{Name: "test"}) + if first.Load() != 0 { + t.Fatal("first handler should have been replaced") + } + if second.Load() != 1 { + t.Fatal("second handler should have been called") + } +} + +// --- Handler registration --- + +func TestRegisterHandler_GetHandler(t *testing.T) { + mb := New() + defer mb.Close() + + called := false + mb.RegisterHandler("telegram", func(msg InboundMessage) error { + called = true + return nil + }) + + handler, ok := mb.GetHandler("telegram") + if !ok { + t.Fatal("expected handler to be registered") + } + _ = handler(InboundMessage{}) + if !called { + t.Fatal("expected handler to be called") + } + + _, ok = mb.GetHandler("nonexistent") + if ok { + t.Fatal("expected no handler for unregistered channel") + } +} + +// --- Concurrent safety --- + +func TestBroadcast_ConcurrentSubscribeUnsubscribe(t *testing.T) { + mb := New() + defer mb.Close() + + var wg sync.WaitGroup + done := make(chan struct{}) + + // Broadcast in a goroutine + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case <-done: + return + default: + mb.Broadcast(Event{Name: "concurrent"}) + } + } + }() + + // Subscribe/unsubscribe rapidly + for i := 0; i < 100; i++ { + mb.Subscribe("rapid", func(e Event) {}) + mb.Unsubscribe("rapid") + } + + close(done) + wg.Wait() + // No panic = success +} + +func TestPublishInbound_ConcurrentProducers(t *testing.T) { + mb := New() + defer mb.Close() + + const n = 100 + var wg sync.WaitGroup + wg.Add(n) + for i := 0; i < n; i++ { + go func() { + defer wg.Done() + mb.TryPublishInbound(InboundMessage{Content: "msg"}) + }() + } + wg.Wait() + // No panic = success +} diff --git a/internal/config/config_load_test.go b/internal/config/config_load_test.go new file mode 100644 index 00000000..ec160925 --- /dev/null +++ b/internal/config/config_load_test.go @@ -0,0 +1,201 @@ +package config + +import ( + "encoding/json" + "os" + "path/filepath" + "testing" +) + +// --- Default --- + +func TestDefault_SensibleDefaults(t *testing.T) { + cfg := Default() + + if cfg.Gateway.Port != 18790 { + t.Fatalf("default port: got %d, want 18790", cfg.Gateway.Port) + } + if cfg.Gateway.RateLimitRPM != 20 { + t.Fatalf("default rate limit: got %d, want 20", cfg.Gateway.RateLimitRPM) + } + if cfg.Agents.Defaults.Provider != "anthropic" { + t.Fatalf("default provider: got %q", cfg.Agents.Defaults.Provider) + } + if cfg.Agents.Defaults.MaxToolIterations != DefaultMaxIterations { + t.Fatalf("default max iterations: got %d", cfg.Agents.Defaults.MaxToolIterations) + } + if cfg.Tools.Web.DuckDuckGo.MaxResults != 5 { + t.Fatalf("default ddg max results: got %d", cfg.Tools.Web.DuckDuckGo.MaxResults) + } +} + +// --- Load with missing file → uses defaults --- + +func TestLoad_MissingFile_UsesDefaults(t *testing.T) { + cfg, err := Load("/nonexistent/path/config.json") + if err != nil { + t.Fatalf("missing file should not error: %v", err) + } + if cfg.Gateway.Port != 18790 { + t.Fatalf("expected default port, got %d", cfg.Gateway.Port) + } +} + +// --- Load with valid JSON5 --- + +func TestLoad_ValidJSON5(t *testing.T) { + dir := t.TempDir() + cfgPath := filepath.Join(dir, "config.json5") + + // JSON5: comments and trailing commas allowed + content := `{ + // custom port + "gateway": { + "port": 9999, + "rate_limit_rpm": 100, + }, + }` + os.WriteFile(cfgPath, []byte(content), 0644) + + cfg, err := Load(cfgPath) + if err != nil { + t.Fatalf("load error: %v", err) + } + if cfg.Gateway.Port != 9999 { + t.Fatalf("port: got %d, want 9999", cfg.Gateway.Port) + } + if cfg.Gateway.RateLimitRPM != 100 { + t.Fatalf("rate limit: got %d, want 100", cfg.Gateway.RateLimitRPM) + } + // Unset fields should retain defaults + if cfg.Agents.Defaults.Provider != "anthropic" { + t.Fatalf("default provider should be preserved: got %q", cfg.Agents.Defaults.Provider) + } +} + +// --- Load with invalid JSON5 → error --- + +func TestLoad_InvalidJSON5(t *testing.T) { + dir := t.TempDir() + cfgPath := filepath.Join(dir, "config.json5") + os.WriteFile(cfgPath, []byte(`{invalid json!!!`), 0644) + + _, err := Load(cfgPath) + if err == nil { + t.Fatal("expected error for invalid JSON5") + } +} + +// --- Env var override precedence --- + +func TestLoad_EnvVarOverrides(t *testing.T) { + dir := t.TempDir() + cfgPath := filepath.Join(dir, "config.json5") + os.WriteFile(cfgPath, []byte(`{"gateway":{"port":8080}}`), 0644) + + // Env override should win + t.Setenv("GOCLAW_PORT", "7777") + + cfg, err := Load(cfgPath) + if err != nil { + t.Fatalf("load error: %v", err) + } + if cfg.Gateway.Port != 7777 { + t.Fatalf("env override: got port %d, want 7777", cfg.Gateway.Port) + } +} + +func TestLoad_EnvVarOverrides_InvalidPort(t *testing.T) { + t.Setenv("GOCLAW_PORT", "not-a-number") + + cfg, err := Load("/nonexistent/path") + if err != nil { + t.Fatalf("load error: %v", err) + } + // Invalid port should keep default + if cfg.Gateway.Port != 18790 { + t.Fatalf("invalid port env should keep default: got %d", cfg.Gateway.Port) + } +} + +// --- Env var for API keys --- + +func TestLoad_EnvVarAPIKeys(t *testing.T) { + t.Setenv("GOCLAW_ANTHROPIC_API_KEY", "sk-test-key") + + cfg, err := Load("/nonexistent/path") + if err != nil { + t.Fatalf("load error: %v", err) + } + if cfg.Providers.Anthropic.APIKey != "sk-test-key" { + t.Fatalf("anthropic key: got %q", cfg.Providers.Anthropic.APIKey) + } +} + +// --- FlexibleStringSlice --- + +func TestFlexibleStringSlice_StringArray(t *testing.T) { + var f FlexibleStringSlice + err := json.Unmarshal([]byte(`["a","b","c"]`), &f) + if err != nil { + t.Fatalf("unmarshal error: %v", err) + } + if len(f) != 3 || f[0] != "a" || f[1] != "b" || f[2] != "c" { + t.Fatalf("got %v", f) + } +} + +func TestFlexibleStringSlice_MixedArray(t *testing.T) { + var f FlexibleStringSlice + // Numbers and strings mixed + err := json.Unmarshal([]byte(`["user1", 12345, "user2"]`), &f) + if err != nil { + t.Fatalf("unmarshal error: %v", err) + } + if len(f) != 3 || f[0] != "user1" || f[1] != "12345" || f[2] != "user2" { + t.Fatalf("got %v", f) + } +} + +func TestFlexibleStringSlice_EmptyArray(t *testing.T) { + var f FlexibleStringSlice + err := json.Unmarshal([]byte(`[]`), &f) + if err != nil { + t.Fatalf("unmarshal error: %v", err) + } + if len(f) != 0 { + t.Fatalf("expected empty, got %v", f) + } +} + +// --- Owner IDs parsing --- + +func TestLoad_OwnerIDsParsing(t *testing.T) { + t.Setenv("GOCLAW_OWNER_IDS", " alice , bob , charlie ") + + cfg, err := Load("/nonexistent/path") + if err != nil { + t.Fatalf("load error: %v", err) + } + if len(cfg.Gateway.OwnerIDs) != 3 { + t.Fatalf("expected 3 owner IDs, got %d: %v", len(cfg.Gateway.OwnerIDs), cfg.Gateway.OwnerIDs) + } + if cfg.Gateway.OwnerIDs[0] != "alice" || cfg.Gateway.OwnerIDs[1] != "bob" || cfg.Gateway.OwnerIDs[2] != "charlie" { + t.Fatalf("owner IDs not trimmed: %v", cfg.Gateway.OwnerIDs) + } +} + +func TestLoad_OwnerIDsEmpty(t *testing.T) { + t.Setenv("GOCLAW_OWNER_IDS", "") + + cfg, err := Load("/nonexistent/path") + if err != nil { + t.Fatalf("load error: %v", err) + } + // Empty string should not produce [""] + for _, id := range cfg.Gateway.OwnerIDs { + if id == "" { + t.Fatal("empty owner ID should not be included") + } + } +} diff --git a/internal/cron/service_execution.go b/internal/cron/service_execution.go index cd50e72c..9f3af89e 100644 --- a/internal/cron/service_execution.go +++ b/internal/cron/service_execution.go @@ -80,8 +80,8 @@ func (cs *Service) RunJob(jobID string, force bool) (bool, string, error) { break } - // Record run log - cs.recordRun(jobID, err, result) + // Record run log (already holding cs.mu via defer above) + cs.recordRunLocked(jobID, err, result) if err != nil { return true, "", err @@ -111,7 +111,11 @@ func (cs *Service) GetRunLog(jobID string, limit int) []RunLogEntry { func (cs *Service) recordRun(jobID string, err error, resultText string) { cs.mu.Lock() defer cs.mu.Unlock() + cs.recordRunLocked(jobID, err, resultText) +} +// recordRunLocked appends a run log entry. Must be called with cs.mu held. +func (cs *Service) recordRunLocked(jobID string, err error, resultText string) { entry := RunLogEntry{ Ts: nowMS(), JobID: jobID, @@ -252,6 +256,9 @@ func (cs *Service) executeJobByID(jobID string) { break } + // Record run log (already holding cs.mu via defer above) + cs.recordRunLocked(jobID, err, result) + cs.saveUnsafe() } diff --git a/internal/cron/service_test.go b/internal/cron/service_test.go new file mode 100644 index 00000000..ed08d03f --- /dev/null +++ b/internal/cron/service_test.go @@ -0,0 +1,357 @@ +package cron + +import ( + "fmt" + "os" + "path/filepath" + "sync/atomic" + "testing" + "time" +) + +// --- Schedule validation --- + +func TestValidateSchedule(t *testing.T) { + cs := NewService("", nil) + + tests := []struct { + name string + sched Schedule + wantErr bool + }{ + {"at_valid", Schedule{Kind: "at", AtMS: ptrInt64(time.Now().Add(time.Hour).UnixMilli())}, false}, + {"at_missing_timestamp", Schedule{Kind: "at"}, true}, + {"every_valid", Schedule{Kind: "every", EveryMS: ptrInt64(5000)}, false}, + {"every_zero_interval", Schedule{Kind: "every", EveryMS: ptrInt64(0)}, true}, + {"every_negative_interval", Schedule{Kind: "every", EveryMS: ptrInt64(-1)}, true}, + {"every_nil_interval", Schedule{Kind: "every"}, true}, + {"cron_valid", Schedule{Kind: "cron", Expr: "*/5 * * * *"}, false}, + {"cron_empty_expr", Schedule{Kind: "cron", Expr: ""}, true}, + {"cron_invalid_expr", Schedule{Kind: "cron", Expr: "bad cron"}, true}, + {"cron_valid_with_tz", Schedule{Kind: "cron", Expr: "0 9 * * *", TZ: "Asia/Saigon"}, false}, + {"cron_invalid_tz", Schedule{Kind: "cron", Expr: "0 9 * * *", TZ: "Invalid/Zone"}, true}, + {"unknown_kind", Schedule{Kind: "invalid"}, true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := cs.validateSchedule(&tt.sched) + if (err != nil) != tt.wantErr { + t.Fatalf("validateSchedule() error = %v, wantErr = %v", err, tt.wantErr) + } + }) + } +} + +// --- computeNextRun --- + +func TestComputeNextRun(t *testing.T) { + cs := NewService("", nil) + now := time.Now().UnixMilli() + + t.Run("at_future", func(t *testing.T) { + future := now + 60000 + sched := Schedule{Kind: "at", AtMS: &future} + next := cs.computeNextRun(&sched, now) + if next == nil || *next != future { + t.Fatalf("expected %d, got %v", future, next) + } + }) + + t.Run("at_past", func(t *testing.T) { + past := now - 60000 + sched := Schedule{Kind: "at", AtMS: &past} + next := cs.computeNextRun(&sched, now) + if next != nil { + t.Fatalf("past at-schedule should return nil, got %d", *next) + } + }) + + t.Run("every_5s", func(t *testing.T) { + interval := int64(5000) + sched := Schedule{Kind: "every", EveryMS: &interval} + next := cs.computeNextRun(&sched, now) + if next == nil { + t.Fatal("expected non-nil next") + } + expected := now + 5000 + if *next != expected { + t.Fatalf("expected %d, got %d", expected, *next) + } + }) + + t.Run("every_nil_interval", func(t *testing.T) { + sched := Schedule{Kind: "every"} + next := cs.computeNextRun(&sched, now) + if next != nil { + t.Fatal("nil interval should return nil") + } + }) + + t.Run("cron_every_minute", func(t *testing.T) { + sched := Schedule{Kind: "cron", Expr: "* * * * *"} + next := cs.computeNextRun(&sched, now) + if next == nil { + t.Fatal("expected non-nil next for every-minute cron") + } + // Should be within next 60 seconds + diff := *next - now + if diff < 0 || diff > 61000 { + t.Fatalf("next run should be within 61s, got diff=%dms", diff) + } + }) + + t.Run("cron_empty_expr", func(t *testing.T) { + sched := Schedule{Kind: "cron", Expr: ""} + next := cs.computeNextRun(&sched, now) + if next != nil { + t.Fatal("empty expr should return nil") + } + }) + + t.Run("unknown_kind", func(t *testing.T) { + sched := Schedule{Kind: "wut"} + next := cs.computeNextRun(&sched, now) + if next != nil { + t.Fatal("unknown kind should return nil") + } + }) +} + +// --- AddJob / RemoveJob / ListJobs / EnableJob --- + +func TestService_CRUD(t *testing.T) { + dir := t.TempDir() + storePath := filepath.Join(dir, "cron.json") + cs := NewService(storePath, nil) + + // Add a job + interval := int64(60000) + job, err := cs.AddJob("test-job", Schedule{Kind: "every", EveryMS: &interval}, "hello", false, "", "", "agent-1") + if err != nil { + t.Fatalf("AddJob error: %v", err) + } + if job.ID == "" { + t.Fatal("job ID should not be empty") + } + if job.Name != "test-job" { + t.Fatalf("job name: got %q", job.Name) + } + if !job.Enabled { + t.Fatal("new job should be enabled") + } + if job.State.NextRunAtMS == nil { + t.Fatal("new job should have NextRunAtMS set") + } + + // List jobs + jobs := cs.ListJobs(false) + if len(jobs) != 1 { + t.Fatalf("expected 1 job, got %d", len(jobs)) + } + + // Disable job + if err := cs.EnableJob(job.ID, false); err != nil { + t.Fatalf("EnableJob error: %v", err) + } + jobs = cs.ListJobs(false) // excludes disabled + if len(jobs) != 0 { + t.Fatalf("expected 0 enabled jobs, got %d", len(jobs)) + } + jobs = cs.ListJobs(true) // includes disabled + if len(jobs) != 1 { + t.Fatalf("expected 1 total job, got %d", len(jobs)) + } + + // Re-enable + if err := cs.EnableJob(job.ID, true); err != nil { + t.Fatalf("EnableJob error: %v", err) + } + + // Remove + if err := cs.RemoveJob(job.ID); err != nil { + t.Fatalf("RemoveJob error: %v", err) + } + jobs = cs.ListJobs(true) + if len(jobs) != 0 { + t.Fatalf("expected 0 jobs after remove, got %d", len(jobs)) + } + + // Verify persisted + if _, err := os.Stat(storePath); os.IsNotExist(err) { + t.Fatal("store file should exist") + } +} + +func TestService_AddJob_InvalidSchedule(t *testing.T) { + cs := NewService("", nil) + _, err := cs.AddJob("bad", Schedule{Kind: "unknown"}, "msg", false, "", "", "") + if err == nil { + t.Fatal("expected error for invalid schedule") + } +} + +func TestService_RemoveJob_NotFound(t *testing.T) { + cs := NewService("", nil) + err := cs.RemoveJob("nonexistent") + if err == nil { + t.Fatal("expected error for nonexistent job") + } +} + +func TestService_EnableJob_NotFound(t *testing.T) { + cs := NewService("", nil) + err := cs.EnableJob("nonexistent", true) + if err == nil { + t.Fatal("expected error for nonexistent job") + } +} + +// --- At-schedule sets DeleteAfterRun --- + +func TestService_AddJob_AtSchedule_DeleteAfterRun(t *testing.T) { + cs := NewService("", nil) + future := time.Now().Add(time.Hour).UnixMilli() + job, err := cs.AddJob("one-shot", Schedule{Kind: "at", AtMS: &future}, "run once", false, "", "", "") + if err != nil { + t.Fatalf("AddJob error: %v", err) + } + if !job.DeleteAfterRun { + t.Fatal("at-schedule should set DeleteAfterRun=true") + } +} + +// --- Job execution callback --- + +func TestService_StartStop_JobExecution(t *testing.T) { + dir := t.TempDir() + storePath := filepath.Join(dir, "cron.json") + + var execCount atomic.Int32 + handler := func(job *Job) (string, error) { + execCount.Add(1) + return "done", nil + } + + cs := NewService(storePath, handler) + + // Add a fast-interval job (every 100ms) — but runLoop ticks every 1s + interval := int64(100) + _, err := cs.AddJob("fast", Schedule{Kind: "every", EveryMS: &interval}, "tick", false, "", "", "") + if err != nil { + t.Fatalf("AddJob error: %v", err) + } + + if err := cs.Start(); err != nil { + t.Fatalf("Start error: %v", err) + } + + // runLoop ticks every 1s, wait enough for at least 1 tick + time.Sleep(1500 * time.Millisecond) + cs.Stop() + + count := execCount.Load() + if count == 0 { + t.Fatal("expected at least 1 job execution") + } +} + +// --- Handler not set → no panic --- + +func TestService_NilHandler_NoPanic(t *testing.T) { + dir := t.TempDir() + storePath := filepath.Join(dir, "cron.json") + + cs := NewService(storePath, nil) // no handler + + interval := int64(100) + cs.AddJob("no-handler", Schedule{Kind: "every", EveryMS: &interval}, "tick", false, "", "", "") + + cs.Start() + time.Sleep(1500 * time.Millisecond) // wait for at least 1 tick + cs.Stop() // should not panic +} + +// --- Job failure with retry --- + +func TestService_JobFailure_Updates_LastError(t *testing.T) { + dir := t.TempDir() + storePath := filepath.Join(dir, "cron.json") + + handler := func(job *Job) (string, error) { + return "", fmt.Errorf("intentional failure") + } + + cs := NewService(storePath, handler) + cs.SetRetryConfig(RetryConfig{MaxRetries: 0}) // no retry + + interval := int64(100) + job, _ := cs.AddJob("failing", Schedule{Kind: "every", EveryMS: &interval}, "fail", false, "", "", "") + + cs.Start() + time.Sleep(1500 * time.Millisecond) // wait for at least 1 tick + cs.Stop() + + // Check last error + found, ok := cs.GetJob(job.ID) + if !ok { + t.Fatal("job should exist") + } + if found.State.LastStatus != "error" { + t.Fatalf("expected last status 'error', got %q", found.State.LastStatus) + } + if found.State.LastError == "" { + t.Fatal("expected non-empty last error") + } +} + +// --- Persistence: save and reload --- + +func TestService_Persistence_Roundtrip(t *testing.T) { + dir := t.TempDir() + storePath := filepath.Join(dir, "cron.json") + + cs1 := NewService(storePath, nil) + interval := int64(60000) + cs1.AddJob("persist-test", Schedule{Kind: "every", EveryMS: &interval}, "msg", false, "", "", "agent-1") + + // New service should load the persisted job + cs2 := NewService(storePath, nil) + cs2.Start() + defer cs2.Stop() + + jobs := cs2.ListJobs(true) + if len(jobs) != 1 { + t.Fatalf("expected 1 persisted job, got %d", len(jobs)) + } + if jobs[0].Name != "persist-test" { + t.Fatalf("job name mismatch: got %q", jobs[0].Name) + } +} + +// --- Run log --- + +func TestService_RunLog_PopulatedByAutoExecution(t *testing.T) { + dir := t.TempDir() + cs := NewService(filepath.Join(dir, "cron.json"), func(job *Job) (string, error) { + return "ok", nil + }) + + interval := int64(100) + job, _ := cs.AddJob("logger", Schedule{Kind: "every", EveryMS: &interval}, "tick", false, "", "", "") + + cs.Start() + time.Sleep(1500 * time.Millisecond) + cs.Stop() + + log := cs.GetRunLog(job.ID, 50) + if len(log) == 0 { + t.Fatal("expected at least 1 run log entry from automatic execution") + } + if log[0].Status != "ok" { + t.Fatalf("expected status 'ok', got %q", log[0].Status) + } +} + +// --- helpers --- + +func ptrInt64(v int64) *int64 { return &v } diff --git a/internal/crypto/aes_test.go b/internal/crypto/aes_test.go new file mode 100644 index 00000000..47eac787 --- /dev/null +++ b/internal/crypto/aes_test.go @@ -0,0 +1,261 @@ +package crypto + +import ( + "encoding/base64" + "encoding/hex" + "strings" + "testing" +) + +// testKey32 is a deterministic 32-byte raw key for tests. +const testKey32 = "01234567890123456789012345678901" // exactly 32 bytes + +func testKeyHex() string { + return hex.EncodeToString([]byte(testKey32)) // 64 hex chars +} + +func testKeyBase64() string { + return base64.StdEncoding.EncodeToString([]byte(testKey32)) // 44 chars ending with = +} + +// --- DeriveKey --- + +func TestDeriveKey_Hex64(t *testing.T) { + k, err := DeriveKey(testKeyHex()) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(k) != 32 { + t.Fatalf("expected 32 bytes, got %d", len(k)) + } +} + +func TestDeriveKey_Base64_44(t *testing.T) { + k, err := DeriveKey(testKeyBase64()) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(k) != 32 { + t.Fatalf("expected 32 bytes, got %d", len(k)) + } +} + +func TestDeriveKey_Raw32(t *testing.T) { + k, err := DeriveKey(testKey32) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if string(k) != testKey32 { + t.Fatalf("raw key mismatch") + } +} + +func TestDeriveKey_InvalidLength(t *testing.T) { + tests := []struct { + name string + key string + }{ + {"too_short", "abc"}, + {"16_bytes", "0123456789abcdef"}, + {"48_bytes", strings.Repeat("a", 48)}, + {"empty", ""}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, err := DeriveKey(tt.key) + if err == nil { + t.Fatalf("expected error for key len %d, got nil", len(tt.key)) + } + }) + } +} + +func TestDeriveKey_Hex64_InvalidHex(t *testing.T) { + // 64 chars but not valid hex → should fall through to raw 32 check, then fail + key := strings.Repeat("zz", 32) // 64 chars, invalid hex + _, err := DeriveKey(key) + if err == nil { + t.Fatal("expected error for invalid hex, got nil") + } +} + +// --- Encrypt with invalid key --- + +func TestEncrypt_InvalidKey(t *testing.T) { + _, err := Encrypt("secret", "too-short-key") + if err == nil { + t.Fatal("expected error for invalid key length") + } +} + +// --- Encrypt / Decrypt roundtrip --- + +func TestEncryptDecrypt_Roundtrip(t *testing.T) { + tests := []struct { + name string + plaintext string + }{ + {"simple_ascii", "my-secret-api-key-12345"}, + {"unicode", "你好世界🌍emoji日本語"}, + {"special_chars", `key="value"&foo=bar<>'"!@#$%`}, + {"long_string", strings.Repeat("abcdefghij", 1000)}, + {"single_char", "x"}, + {"whitespace", " \t\n "}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + enc, err := Encrypt(tt.plaintext, testKey32) + if err != nil { + t.Fatalf("encrypt failed: %v", err) + } + if !strings.HasPrefix(enc, prefix) { + t.Fatalf("encrypted value missing prefix, got: %s", enc[:20]) + } + + dec, err := Decrypt(enc, testKey32) + if err != nil { + t.Fatalf("decrypt failed: %v", err) + } + if dec != tt.plaintext { + t.Fatalf("roundtrip mismatch: got %q, want %q", dec, tt.plaintext) + } + }) + } +} + +func TestEncryptDecrypt_EmptyPlaintext(t *testing.T) { + enc, err := Encrypt("", testKey32) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if enc != "" { + t.Fatalf("expected empty string, got %q", enc) + } +} + +func TestEncryptDecrypt_EmptyKey(t *testing.T) { + // Empty key → plaintext returned unchanged for both encrypt and decrypt + enc, err := Encrypt("secret", "") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if enc != "secret" { + t.Fatalf("expected plaintext passthrough, got %q", enc) + } + + dec, err := Decrypt("aes-gcm:someciphertext", "") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if dec != "aes-gcm:someciphertext" { + t.Fatalf("expected ciphertext passthrough, got %q", dec) + } +} + +// --- Nonce uniqueness: same plaintext + key → different ciphertexts --- + +func TestEncrypt_NonceUniqueness(t *testing.T) { + const plaintext = "identical-plaintext" + enc1, _ := Encrypt(plaintext, testKey32) + enc2, _ := Encrypt(plaintext, testKey32) + if enc1 == enc2 { + t.Fatal("two encryptions of same plaintext must produce different ciphertext (unique nonce)") + } +} + +// --- Backward compatibility: decrypt unencrypted string returns as-is --- + +func TestDecrypt_BackwardCompat_PlainText(t *testing.T) { + plain := "sk-ant-not-encrypted-just-raw" + dec, err := Decrypt(plain, testKey32) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if dec != plain { + t.Fatalf("expected passthrough for unencrypted value, got %q", dec) + } +} + +// --- Error cases --- + +func TestDecrypt_CorruptedCiphertext(t *testing.T) { + // Valid prefix but garbage base64 → should return as-is (line 68: treat as plain text) + garbage := prefix + "!!!not-base64!!!" + dec, err := Decrypt(garbage, testKey32) + if err != nil { + t.Fatalf("unexpected error for garbage base64: %v", err) + } + if dec != garbage { + t.Fatalf("expected passthrough for invalid base64, got %q", dec) + } +} + +func TestDecrypt_WrongKey(t *testing.T) { + enc, _ := Encrypt("secret", testKey32) + + otherKey := "98765432109876543210987654321098" // different 32-byte key + _, err := Decrypt(enc, otherKey) + if err == nil { + t.Fatal("expected error for wrong key, got nil") + } + if !strings.Contains(err.Error(), "decrypt failed") { + t.Fatalf("expected 'decrypt failed' error, got: %v", err) + } +} + +func TestDecrypt_TooShortCiphertext(t *testing.T) { + // Valid prefix + valid base64 but too short for nonce → return as-is + short := prefix + base64.StdEncoding.EncodeToString([]byte("x")) + dec, err := Decrypt(short, testKey32) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if dec != short { + t.Fatalf("expected passthrough for too-short ciphertext, got %q", dec) + } +} + +// --- IsEncrypted --- + +func TestIsEncrypted(t *testing.T) { + if !IsEncrypted("aes-gcm:abc123") { + t.Fatal("expected true for prefixed value") + } + if IsEncrypted("plain-text-value") { + t.Fatal("expected false for unprefixed value") + } + if IsEncrypted("") { + t.Fatal("expected false for empty string") + } +} + +// --- Cross-key-format roundtrip: encrypt with one format, decrypt with another --- + +func TestEncryptDecrypt_CrossKeyFormats(t *testing.T) { + // All three key formats represent the same 32-byte key + keyRaw := testKey32 + keyHex := testKeyHex() + keyB64 := testKeyBase64() + + plaintext := "cross-format-test-value" + + // Encrypt with raw, decrypt with hex + enc, _ := Encrypt(plaintext, keyRaw) + dec, err := Decrypt(enc, keyHex) + if err != nil { + t.Fatalf("hex decrypt failed: %v", err) + } + if dec != plaintext { + t.Fatalf("cross-format mismatch: raw→hex") + } + + // Encrypt with hex, decrypt with base64 + enc, _ = Encrypt(plaintext, keyHex) + dec, err = Decrypt(enc, keyB64) + if err != nil { + t.Fatalf("base64 decrypt failed: %v", err) + } + if dec != plaintext { + t.Fatalf("cross-format mismatch: hex→base64") + } +} diff --git a/internal/permissions/policy_test.go b/internal/permissions/policy_test.go new file mode 100644 index 00000000..783ae7d1 --- /dev/null +++ b/internal/permissions/policy_test.go @@ -0,0 +1,280 @@ +package permissions + +import ( + "testing" + + "github.com/nextlevelbuilder/goclaw/pkg/protocol" +) + +// --- Role hierarchy --- + +func TestRoleLevel_Ordering(t *testing.T) { + // Owner > Admin > Operator > Viewer > unknown + levels := []struct { + role Role + level int + }{ + {RoleOwner, 4}, + {RoleAdmin, 3}, + {RoleOperator, 2}, + {RoleViewer, 1}, + {Role("unknown"), 0}, + {Role(""), 0}, + } + for _, tt := range levels { + t.Run(string(tt.role), func(t *testing.T) { + got := roleLevel(tt.role) + if got != tt.level { + t.Fatalf("roleLevel(%q) = %d, want %d", tt.role, got, tt.level) + } + }) + } +} + +func TestHasMinRole(t *testing.T) { + tests := []struct { + name string + role Role + required Role + want bool + }{ + {"owner_meets_admin", RoleOwner, RoleAdmin, true}, + {"admin_meets_admin", RoleAdmin, RoleAdmin, true}, + {"operator_fails_admin", RoleOperator, RoleAdmin, false}, + {"viewer_fails_operator", RoleViewer, RoleOperator, false}, + {"operator_meets_viewer", RoleOperator, RoleViewer, true}, + {"admin_meets_viewer", RoleAdmin, RoleViewer, true}, + {"viewer_meets_viewer", RoleViewer, RoleViewer, true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := HasMinRole(tt.role, tt.required) + if got != tt.want { + t.Fatalf("HasMinRole(%q, %q) = %v, want %v", tt.role, tt.required, got, tt.want) + } + }) + } +} + +// --- RoleFromScopes --- + +func TestRoleFromScopes(t *testing.T) { + tests := []struct { + name string + scopes []Scope + want Role + }{ + {"admin_scope", []Scope{ScopeAdmin}, RoleAdmin}, + {"admin_overrides_read", []Scope{ScopeRead, ScopeAdmin}, RoleAdmin}, + {"write_is_operator", []Scope{ScopeWrite}, RoleOperator}, + {"approvals_is_operator", []Scope{ScopeApprovals}, RoleOperator}, + {"pairing_is_operator", []Scope{ScopePairing}, RoleOperator}, + {"read_is_viewer", []Scope{ScopeRead}, RoleViewer}, + {"empty_scopes", []Scope{}, RoleViewer}, + {"nil_scopes", nil, RoleViewer}, + {"provision_only_is_viewer", []Scope{ScopeProvision}, RoleViewer}, + {"read_and_write", []Scope{ScopeRead, ScopeWrite}, RoleOperator}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := RoleFromScopes(tt.scopes) + if got != tt.want { + t.Fatalf("RoleFromScopes(%v) = %q, want %q", tt.scopes, got, tt.want) + } + }) + } +} + +// --- CanAccess: role-based method access --- + +func TestCanAccess_AdminMethods(t *testing.T) { + pe := NewPolicyEngine(nil) + adminMethods := []string{ + protocol.MethodConfigApply, + protocol.MethodAgentsCreate, + protocol.MethodAgentsDelete, + protocol.MethodAPIKeysCreate, + protocol.MethodTeamsCreate, + } + for _, method := range adminMethods { + t.Run(method, func(t *testing.T) { + if !pe.CanAccess(RoleAdmin, method) { + t.Fatalf("admin should access %s", method) + } + if !pe.CanAccess(RoleOwner, method) { + t.Fatalf("owner should access %s", method) + } + if pe.CanAccess(RoleOperator, method) { + t.Fatalf("operator should NOT access %s", method) + } + if pe.CanAccess(RoleViewer, method) { + t.Fatalf("viewer should NOT access %s", method) + } + }) + } +} + +func TestCanAccess_WriteMethods(t *testing.T) { + pe := NewPolicyEngine(nil) + writeMethods := []string{ + protocol.MethodChatSend, + protocol.MethodSessionsDelete, + protocol.MethodCronCreate, + } + for _, method := range writeMethods { + t.Run(method, func(t *testing.T) { + if !pe.CanAccess(RoleOperator, method) { + t.Fatalf("operator should access %s", method) + } + if !pe.CanAccess(RoleAdmin, method) { + t.Fatalf("admin should access %s", method) + } + if pe.CanAccess(RoleViewer, method) { + t.Fatalf("viewer should NOT access write method %s", method) + } + }) + } +} + +func TestCanAccess_ReadMethods_AnyRole(t *testing.T) { + pe := NewPolicyEngine(nil) + // A method not in admin or write lists → defaults to viewer + readMethod := "sessions.list" // not in admin/write lists + for _, role := range []Role{RoleViewer, RoleOperator, RoleAdmin, RoleOwner} { + if !pe.CanAccess(role, readMethod) { + t.Fatalf("%s should access read method %s", role, readMethod) + } + } +} + +func TestCanAccess_UnknownMethod_DefaultsToViewer(t *testing.T) { + pe := NewPolicyEngine(nil) + // Unknown method defaults to viewer role requirement + if !pe.CanAccess(RoleViewer, "totally.unknown.method") { + t.Fatal("viewer should access unknown method (defaults to viewer)") + } +} + +// --- CanAccessWithScopes: scope-based method access --- + +func TestCanAccessWithScopes(t *testing.T) { + pe := NewPolicyEngine(nil) + + tests := []struct { + name string + scopes []Scope + method string + want bool + }{ + // Admin method requires ScopeAdmin + {"admin_scope_for_admin_method", []Scope{ScopeAdmin}, protocol.MethodAgentsCreate, true}, + {"read_scope_for_admin_method", []Scope{ScopeRead}, protocol.MethodAgentsCreate, false}, + {"write_scope_for_admin_method", []Scope{ScopeWrite}, protocol.MethodAgentsCreate, false}, + + // Approvals method requires ScopeApprovals or ScopeAdmin + {"approvals_scope_for_approvals", []Scope{ScopeApprovals}, "approvals.list", true}, + {"admin_scope_for_approvals", []Scope{ScopeAdmin}, "approvals.list", true}, + {"read_scope_for_approvals", []Scope{ScopeRead}, "approvals.list", false}, + + // Write method requires ScopeWrite or ScopeAdmin + {"write_scope_for_chat", []Scope{ScopeWrite}, protocol.MethodChatSend, true}, + {"admin_scope_for_chat", []Scope{ScopeAdmin}, protocol.MethodChatSend, true}, + {"read_scope_for_chat", []Scope{ScopeRead}, protocol.MethodChatSend, false}, + + // Read method allows ScopeRead, ScopeWrite, or ScopeAdmin + {"read_scope_for_read_method", []Scope{ScopeRead}, "sessions.list", true}, + {"write_scope_for_read_method", []Scope{ScopeWrite}, "sessions.list", true}, + {"admin_scope_for_read_method", []Scope{ScopeAdmin}, "sessions.list", true}, + + // Empty scopes + {"empty_scopes", []Scope{}, protocol.MethodAgentsCreate, false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := pe.CanAccessWithScopes(tt.scopes, tt.method) + if got != tt.want { + t.Fatalf("CanAccessWithScopes(%v, %q) = %v, want %v", tt.scopes, tt.method, got, tt.want) + } + }) + } +} + +// --- IsOwner --- + +func TestIsOwner(t *testing.T) { + pe := NewPolicyEngine([]string{"alice", "bob"}) + + if !pe.IsOwner("alice") { + t.Fatal("alice should be owner") + } + if !pe.IsOwner("bob") { + t.Fatal("bob should be owner") + } + if pe.IsOwner("charlie") { + t.Fatal("charlie should not be owner") + } + if pe.IsOwner("") { + t.Fatal("empty string should not be owner") + } +} + +func TestIsOwner_EmptyList(t *testing.T) { + pe := NewPolicyEngine(nil) + if pe.IsOwner("anyone") { + t.Fatal("no one should be owner with empty list") + } +} + +// --- ValidScope --- + +func TestValidScope(t *testing.T) { + for scope := range AllScopes { + if !ValidScope(string(scope)) { + t.Fatalf("expected %q to be valid", scope) + } + } + if ValidScope("nonexistent.scope") { + t.Fatal("expected invalid scope to be rejected") + } + if ValidScope("") { + t.Fatal("expected empty scope to be rejected") + } +} + +// --- MethodScopes: verify pairing/approvals special routes --- + +func TestMethodScopes_PairingMethod(t *testing.T) { + scopes := MethodScopes("pairing.request") + if len(scopes) != 2 { + t.Fatalf("expected 2 scopes for pairing method, got %d", len(scopes)) + } + // Should require ScopePairing or ScopeAdmin + hasPairing, hasAdmin := false, false + for _, s := range scopes { + if s == ScopePairing { + hasPairing = true + } + if s == ScopeAdmin { + hasAdmin = true + } + } + if !hasPairing || !hasAdmin { + t.Fatalf("pairing method should require [pairing, admin], got %v", scopes) + } +} + +func TestMethodScopes_ApprovalMethod(t *testing.T) { + scopes := MethodScopes("approvals.list") + hasScopeApprovals, hasAdmin := false, false + for _, s := range scopes { + if s == ScopeApprovals { + hasScopeApprovals = true + } + if s == ScopeAdmin { + hasAdmin = true + } + } + if !hasScopeApprovals || !hasAdmin { + t.Fatalf("approvals method should require [approvals, admin], got %v", scopes) + } +} diff --git a/internal/providers/retry_test.go b/internal/providers/retry_test.go new file mode 100644 index 00000000..bd95a410 --- /dev/null +++ b/internal/providers/retry_test.go @@ -0,0 +1,330 @@ +package providers + +import ( + "context" + "errors" + "fmt" + "sync/atomic" + "testing" + "time" +) + +// --- IsRetryableError --- + +func TestIsRetryableError(t *testing.T) { + tests := []struct { + name string + err error + want bool + }{ + {"nil", nil, false}, + {"http_429_rate_limit", &HTTPError{Status: 429}, true}, + {"http_500_server_error", &HTTPError{Status: 500}, true}, + {"http_502_bad_gateway", &HTTPError{Status: 502}, true}, + {"http_503_unavailable", &HTTPError{Status: 503}, true}, + {"http_504_timeout", &HTTPError{Status: 504}, true}, + {"http_400_bad_request", &HTTPError{Status: 400}, false}, + {"http_401_unauthorized", &HTTPError{Status: 401}, false}, + {"http_403_forbidden", &HTTPError{Status: 403}, false}, + {"http_404_not_found", &HTTPError{Status: 404}, false}, + {"connection_reset", errors.New("connection reset by peer"), true}, + {"broken_pipe", errors.New("write: broken pipe"), true}, + {"eof", errors.New("unexpected EOF"), true}, + {"timeout_string", errors.New("i/o timeout"), true}, + {"generic_error", errors.New("something went wrong"), false}, + {"wrapped_retryable", fmt.Errorf("provider: %w", &HTTPError{Status: 429}), true}, + {"wrapped_non_retryable", fmt.Errorf("provider: %w", &HTTPError{Status: 400}), false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := IsRetryableError(tt.err) + if got != tt.want { + t.Fatalf("IsRetryableError(%v) = %v, want %v", tt.err, got, tt.want) + } + }) + } +} + +// --- computeDelay --- + +func TestComputeDelay_ExponentialBackoff(t *testing.T) { + cfg := RetryConfig{ + MinDelay: 100 * time.Millisecond, + MaxDelay: 10 * time.Second, + Jitter: 0, // no jitter for deterministic test + } + err := &HTTPError{Status: 500} + + // attempt 1: 100ms * 2^0 = 100ms + d1 := computeDelay(cfg, 1, err) + if d1 != 100*time.Millisecond { + t.Fatalf("attempt 1: got %v, want 100ms", d1) + } + + // attempt 2: 100ms * 2^1 = 200ms + d2 := computeDelay(cfg, 2, err) + if d2 != 200*time.Millisecond { + t.Fatalf("attempt 2: got %v, want 200ms", d2) + } + + // attempt 3: 100ms * 2^2 = 400ms + d3 := computeDelay(cfg, 3, err) + if d3 != 400*time.Millisecond { + t.Fatalf("attempt 3: got %v, want 400ms", d3) + } + + // attempt 4: 100ms * 2^3 = 800ms + d4 := computeDelay(cfg, 4, err) + if d4 != 800*time.Millisecond { + t.Fatalf("attempt 4: got %v, want 800ms", d4) + } +} + +func TestComputeDelay_CappedAtMaxDelay(t *testing.T) { + cfg := RetryConfig{ + MinDelay: 1 * time.Second, + MaxDelay: 5 * time.Second, + Jitter: 0, + } + err := &HTTPError{Status: 500} + + // attempt 10: 1s * 2^9 = 512s → capped at 5s + d := computeDelay(cfg, 10, err) + if d != 5*time.Second { + t.Fatalf("attempt 10: got %v, want 5s (capped)", d) + } +} + +func TestComputeDelay_JitterRange(t *testing.T) { + cfg := RetryConfig{ + MinDelay: 1 * time.Second, + MaxDelay: 30 * time.Second, + Jitter: 0.25, // ±25% + } + err := &HTTPError{Status: 500} + + // attempt 1: base = 1s, jitter ±25% → [750ms, 1250ms] + min := 750 * time.Millisecond + max := 1250 * time.Millisecond + + for i := 0; i < 100; i++ { + d := computeDelay(cfg, 1, err) + if d < min || d > max { + t.Fatalf("jitter out of range: got %v, want [%v, %v]", d, min, max) + } + } +} + +func TestComputeDelay_NeverNegative(t *testing.T) { + cfg := RetryConfig{ + MinDelay: 10 * time.Millisecond, + MaxDelay: 100 * time.Millisecond, + Jitter: 0.9, // extreme jitter + } + err := &HTTPError{Status: 500} + + for i := 0; i < 200; i++ { + d := computeDelay(cfg, 1, err) + if d < 0 { + t.Fatalf("negative delay: %v", d) + } + } +} + +func TestComputeDelay_RetryAfterOverride(t *testing.T) { + cfg := RetryConfig{ + MinDelay: 100 * time.Millisecond, + MaxDelay: 30 * time.Second, + Jitter: 0.1, + } + // HTTPError with RetryAfter should override computed delay + err := &HTTPError{Status: 429, RetryAfter: 42 * time.Second} + + d := computeDelay(cfg, 1, err) + if d != 42*time.Second { + t.Fatalf("expected Retry-After override: got %v, want 42s", d) + } +} + +// --- ParseRetryAfter --- + +func TestParseRetryAfter(t *testing.T) { + tests := []struct { + name string + value string + want time.Duration + }{ + {"empty", "", 0}, + {"integer_seconds", "30", 30 * time.Second}, + {"zero", "0", 0}, + {"negative_int", "-5", -5 * time.Second}, // strconv.Atoi succeeds → returns negative duration (caller should clamp) + {"non_numeric", "abc", 0}, // neither int nor date + {"float", "1.5", 0}, // not a valid int + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := ParseRetryAfter(tt.value) + if got != tt.want { + t.Fatalf("ParseRetryAfter(%q) = %v, want %v", tt.value, got, tt.want) + } + }) + } +} + +func TestParseRetryAfter_RFC1123(t *testing.T) { + // Use a future date so time.Until() returns positive + future := time.Now().Add(60 * time.Second).UTC().Format(time.RFC1123) + got := ParseRetryAfter(future) + // Should be roughly 60s (allow ±5s tolerance for test execution time) + if got < 55*time.Second || got > 65*time.Second { + t.Fatalf("RFC1123 parse: got %v, want ~60s", got) + } +} + +func TestParseRetryAfter_PastDate(t *testing.T) { + past := time.Now().Add(-60 * time.Second).UTC().Format(time.RFC1123) + got := ParseRetryAfter(past) + if got != 0 { + t.Fatalf("past date should return 0, got %v", got) + } +} + +// --- RetryDo --- + +func TestRetryDo_SuccessOnFirstAttempt(t *testing.T) { + cfg := RetryConfig{Attempts: 3, MinDelay: time.Millisecond} + var calls int + result, err := RetryDo(context.Background(), cfg, func() (string, error) { + calls++ + return "ok", nil + }) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if result != "ok" { + t.Fatalf("got %q, want %q", result, "ok") + } + if calls != 1 { + t.Fatalf("expected 1 call, got %d", calls) + } +} + +func TestRetryDo_SuccessAfterRetries(t *testing.T) { + cfg := RetryConfig{Attempts: 3, MinDelay: time.Millisecond, MaxDelay: 10 * time.Millisecond} + var calls int + result, err := RetryDo(context.Background(), cfg, func() (string, error) { + calls++ + if calls < 3 { + return "", &HTTPError{Status: 500, Body: "server error"} + } + return "recovered", nil + }) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if result != "recovered" { + t.Fatalf("got %q, want %q", result, "recovered") + } + if calls != 3 { + t.Fatalf("expected 3 calls, got %d", calls) + } +} + +func TestRetryDo_NonRetryableError_NoRetry(t *testing.T) { + cfg := RetryConfig{Attempts: 5, MinDelay: time.Millisecond} + var calls int + _, err := RetryDo(context.Background(), cfg, func() (string, error) { + calls++ + return "", &HTTPError{Status: 400, Body: "bad request"} + }) + if err == nil { + t.Fatal("expected error") + } + if calls != 1 { + t.Fatalf("non-retryable error should not retry: got %d calls, want 1", calls) + } +} + +func TestRetryDo_MaxAttemptsExhausted(t *testing.T) { + cfg := RetryConfig{Attempts: 3, MinDelay: time.Millisecond, MaxDelay: 10 * time.Millisecond} + var calls int + _, err := RetryDo(context.Background(), cfg, func() (string, error) { + calls++ + return "", &HTTPError{Status: 503, Body: "unavailable"} + }) + if err == nil { + t.Fatal("expected error after exhausting retries") + } + if calls != 3 { + t.Fatalf("expected 3 attempts, got %d", calls) + } +} + +func TestRetryDo_ContextCancellation(t *testing.T) { + cfg := RetryConfig{Attempts: 10, MinDelay: 5 * time.Second, MaxDelay: 5 * time.Second} + ctx, cancel := context.WithCancel(context.Background()) + + go func() { + time.Sleep(50 * time.Millisecond) + cancel() + }() + + start := time.Now() + _, err := RetryDo(ctx, cfg, func() (string, error) { + return "", &HTTPError{Status: 500} + }) + elapsed := time.Since(start) + + if !errors.Is(err, context.Canceled) { + t.Fatalf("expected context.Canceled, got: %v", err) + } + // Should have been cancelled quickly, not waited for 5s backoff + if elapsed > 2*time.Second { + t.Fatalf("context cancellation took too long: %v", elapsed) + } +} + +func TestRetryDo_HookCalledOnRetry(t *testing.T) { + cfg := RetryConfig{Attempts: 3, MinDelay: time.Millisecond, MaxDelay: 10 * time.Millisecond} + + var hookCalls atomic.Int32 + ctx := WithRetryHook(context.Background(), func(attempt, maxAttempts int, err error) { + hookCalls.Add(1) + }) + + RetryDo(ctx, cfg, func() (string, error) { + return "", &HTTPError{Status: 500} + }) + + // Hook called before each retry (not the first attempt, not the last failure) + // With 3 attempts: attempt 1 fails → hook → attempt 2 fails → hook → attempt 3 fails → done + if got := hookCalls.Load(); got != 2 { + t.Fatalf("expected 2 hook calls, got %d", got) + } +} + +func TestRetryDo_ZeroAttempts_DefaultsToOne(t *testing.T) { + cfg := RetryConfig{Attempts: 0} + var calls int + _, err := RetryDo(context.Background(), cfg, func() (string, error) { + calls++ + return "", &HTTPError{Status: 500} + }) + if err == nil { + t.Fatal("expected error") + } + if calls != 1 { + t.Fatalf("zero attempts should default to 1: got %d calls", calls) + } +} + +// --- HTTPError --- + +func TestHTTPError_ErrorString(t *testing.T) { + err := &HTTPError{Status: 429, Body: "rate limited"} + got := err.Error() + want := "HTTP 429: rate limited" + if got != want { + t.Fatalf("got %q, want %q", got, want) + } +} diff --git a/internal/scheduler/scheduler_queue_test.go b/internal/scheduler/scheduler_queue_test.go new file mode 100644 index 00000000..22814e38 --- /dev/null +++ b/internal/scheduler/scheduler_queue_test.go @@ -0,0 +1,466 @@ +package scheduler + +import ( + "context" + "errors" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/nextlevelbuilder/goclaw/internal/agent" +) + +// mockRunFn creates a run function that completes after a delay. +func mockRunFn(delay time.Duration) RunFunc { + return func(ctx context.Context, req agent.RunRequest) (*agent.RunResult, error) { + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(delay): + return &agent.RunResult{Content: "done:" + req.RunID}, nil + } + } +} + +// fastRunFn completes immediately. +func fastRunFn() RunFunc { + return func(ctx context.Context, req agent.RunRequest) (*agent.RunResult, error) { + return &agent.RunResult{Content: "ok"}, nil + } +} + +// --- Draining mode rejects new requests --- + +func TestScheduler_MarkDraining(t *testing.T) { + sched := NewScheduler(nil, DefaultQueueConfig(), fastRunFn()) + defer sched.Stop() + + sched.MarkDraining() + + ch := sched.Schedule(context.Background(), LaneMain, agent.RunRequest{ + SessionKey: "agent:a1:s1", + RunID: "run-1", + }) + + outcome := <-ch + if !errors.Is(outcome.Err, ErrGatewayDraining) { + t.Fatalf("expected ErrGatewayDraining, got: %v", outcome.Err) + } +} + +// --- DropNew policy: full queue rejects incoming --- + +func TestSessionQueue_DropNewPolicy(t *testing.T) { + cfg := QueueConfig{ + Mode: QueueModeQueue, + Cap: 2, + Drop: DropNew, + DebounceMs: 0, + MaxConcurrent: 1, + } + + // Use a slow run to keep the queue occupied + blockCh := make(chan struct{}) + runFn := func(ctx context.Context, req agent.RunRequest) (*agent.RunResult, error) { + <-blockCh + return &agent.RunResult{}, nil + } + + laneMgr := NewLaneManager([]LaneConfig{{Name: LaneMain, Concurrency: 10}}) + sq := NewSessionQueue("test-session", LaneMain, cfg, laneMgr, runFn) + + // Enqueue 3 requests: 1 active + 2 in queue = at capacity + ctx := context.Background() + sq.Enqueue(ctx, agent.RunRequest{RunID: "r1", SessionKey: "s"}) + time.Sleep(10 * time.Millisecond) // let r1 start + sq.Enqueue(ctx, agent.RunRequest{RunID: "r2", SessionKey: "s"}) + sq.Enqueue(ctx, agent.RunRequest{RunID: "r3", SessionKey: "s"}) + + // Queue is full (cap=2). Next one should be rejected. + ch := sq.Enqueue(ctx, agent.RunRequest{RunID: "r4", SessionKey: "s"}) + + outcome := <-ch + if !errors.Is(outcome.Err, ErrQueueFull) { + t.Fatalf("expected ErrQueueFull, got: %v", outcome.Err) + } + + close(blockCh) // unblock +} + +// --- DropOld policy: full queue drops oldest --- + +func TestSessionQueue_DropOldPolicy(t *testing.T) { + cfg := QueueConfig{ + Mode: QueueModeQueue, + Cap: 2, + Drop: DropOld, + DebounceMs: 0, + MaxConcurrent: 1, + } + + blockCh := make(chan struct{}) + runFn := func(ctx context.Context, req agent.RunRequest) (*agent.RunResult, error) { + <-blockCh + return &agent.RunResult{}, nil + } + + laneMgr := NewLaneManager([]LaneConfig{{Name: LaneMain, Concurrency: 10}}) + sq := NewSessionQueue("test-session", LaneMain, cfg, laneMgr, runFn) + + ctx := context.Background() + sq.Enqueue(ctx, agent.RunRequest{RunID: "r1", SessionKey: "s"}) + time.Sleep(10 * time.Millisecond) + + ch2 := sq.Enqueue(ctx, agent.RunRequest{RunID: "r2", SessionKey: "s"}) + sq.Enqueue(ctx, agent.RunRequest{RunID: "r3", SessionKey: "s"}) + + // Queue is full. Adding r4 should drop r2 (oldest queued). + sq.Enqueue(ctx, agent.RunRequest{RunID: "r4", SessionKey: "s"}) + + // r2 should have been dropped + outcome := <-ch2 + if !errors.Is(outcome.Err, ErrQueueDropped) { + t.Fatalf("expected ErrQueueDropped for r2, got: %v", outcome.Err) + } + + close(blockCh) +} + +// --- Adaptive throttle: reduces concurrency near 60% context usage --- + +func TestSessionQueue_AdaptiveThrottle(t *testing.T) { + cfg := QueueConfig{ + Mode: QueueModeQueue, + Cap: 10, + Drop: DropOld, + DebounceMs: 0, + MaxConcurrent: 5, + } + + laneMgr := NewLaneManager([]LaneConfig{{Name: LaneMain, Concurrency: 20}}) + sq := NewSessionQueue("test-session", LaneMain, cfg, laneMgr, fastRunFn()) + + // Without token estimate → effectiveMaxConcurrent = 5 + sq.mu.Lock() + got := sq.effectiveMaxConcurrent() + sq.mu.Unlock() + if got != 5 { + t.Fatalf("without estimate: got %d, want 5", got) + } + + // With token estimate under threshold (50%) → still 5 + sq.tokenEstimateFn = func(key string) (int, int) { + return 50000, 100000 // 50% + } + sq.mu.Lock() + got = sq.effectiveMaxConcurrent() + sq.mu.Unlock() + if got != 5 { + t.Fatalf("under threshold: got %d, want 5", got) + } + + // At 60% threshold → drops to 1 + sq.tokenEstimateFn = func(key string) (int, int) { + return 60000, 100000 // 60% + } + sq.mu.Lock() + got = sq.effectiveMaxConcurrent() + sq.mu.Unlock() + if got != 1 { + t.Fatalf("at threshold: got %d, want 1", got) + } + + // Above threshold → still 1 + sq.tokenEstimateFn = func(key string) (int, int) { + return 80000, 100000 // 80% + } + sq.mu.Lock() + got = sq.effectiveMaxConcurrent() + sq.mu.Unlock() + if got != 1 { + t.Fatalf("above threshold: got %d, want 1", got) + } +} + +// --- Generation-based stale completion filtering --- + +func TestSessionQueue_StaleCompletion_AfterReset(t *testing.T) { + var runCount atomic.Int32 + blockCh := make(chan struct{}) + + runFn := func(ctx context.Context, req agent.RunRequest) (*agent.RunResult, error) { + runCount.Add(1) + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-blockCh: + return &agent.RunResult{Content: "from-" + req.RunID}, nil + } + } + + cfg := QueueConfig{Mode: QueueModeQueue, Cap: 10, DebounceMs: 0, MaxConcurrent: 1} + laneMgr := NewLaneManager([]LaneConfig{{Name: LaneMain, Concurrency: 10}}) + sq := NewSessionQueue("test", LaneMain, cfg, laneMgr, runFn) + + // Start run r1 + ch1 := sq.Enqueue(context.Background(), agent.RunRequest{RunID: "r1", SessionKey: "test"}) + time.Sleep(20 * time.Millisecond) // let r1 start + + // Reset bumps generation, cancels r1 + sq.Reset() + + // r1's context is cancelled → it should complete with error + close(blockCh) + outcome := <-ch1 + // r1 was either cancelled or completed, but the key test is the generation check + _ = outcome + + // After reset, new runs should work normally + sq2ch := sq.Enqueue(context.Background(), agent.RunRequest{RunID: "r2", SessionKey: "test"}) + outcome2 := <-sq2ch + if outcome2.Err != nil { + t.Fatalf("post-reset run should succeed: %v", outcome2.Err) + } +} + +// --- CancelAll sets abort cutoff for stale messages --- + +func TestSessionQueue_CancelAll_StaleMessages(t *testing.T) { + blockCh := make(chan struct{}) + runFn := func(ctx context.Context, req agent.RunRequest) (*agent.RunResult, error) { + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-blockCh: + return &agent.RunResult{}, nil + } + } + + cfg := QueueConfig{Mode: QueueModeQueue, Cap: 10, DebounceMs: 0, MaxConcurrent: 1} + laneMgr := NewLaneManager([]LaneConfig{{Name: LaneMain, Concurrency: 10}}) + sq := NewSessionQueue("test", LaneMain, cfg, laneMgr, runFn) + + ctx := context.Background() + sq.Enqueue(ctx, agent.RunRequest{RunID: "r1", SessionKey: "test"}) + time.Sleep(10 * time.Millisecond) + + // Queue r2 before cancellation + ch2 := sq.Enqueue(ctx, agent.RunRequest{RunID: "r2", SessionKey: "test"}) + + // CancelAll → sets abort cutoff → r2 was queued before cutoff → stale + sq.CancelAll() + + outcome := <-ch2 + // r2 was in queue when CancelAll drained it → drainQueue sends context.Canceled + if !errors.Is(outcome.Err, context.Canceled) { + t.Fatalf("expected context.Canceled for drained message, got: %v", outcome.Err) + } + + close(blockCh) +} + +// --- Lane concurrency enforcement --- + +func TestLane_ConcurrencyEnforcement(t *testing.T) { + lane := NewLane("test", 2) + defer lane.Stop() + + var maxConcurrent atomic.Int32 + var current atomic.Int32 + var wg sync.WaitGroup + + for i := 0; i < 10; i++ { + wg.Add(1) + err := lane.Submit(context.Background(), func() { + defer wg.Done() + cur := current.Add(1) + // Track max concurrent + for { + old := maxConcurrent.Load() + if cur <= old || maxConcurrent.CompareAndSwap(old, cur) { + break + } + } + time.Sleep(10 * time.Millisecond) + current.Add(-1) + }) + if err != nil { + wg.Done() + } + } + + wg.Wait() + + if got := maxConcurrent.Load(); got > 2 { + t.Fatalf("max concurrent exceeded limit: got %d, want ≤2", got) + } +} + +// --- Lane Submit with cancelled context --- + +func TestLane_Submit_CancelledContext(t *testing.T) { + lane := NewLane("test", 1) + defer lane.Stop() + + // Fill the single slot + blockCh := make(chan struct{}) + lane.Submit(context.Background(), func() { + <-blockCh + }) + + // Submit with already-cancelled context + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + err := lane.Submit(ctx, func() { + t.Fatal("should not execute") + }) + if err == nil { + t.Fatal("expected error for cancelled context") + } + + close(blockCh) +} + +// --- LaneManager fallback to main --- + +func TestLaneManager_FallbackToMain(t *testing.T) { + lm := NewLaneManager([]LaneConfig{ + {Name: LaneMain, Concurrency: 5}, + }) + + // Known lane + if lm.Get(LaneMain) == nil { + t.Fatal("main lane should exist") + } + + // Unknown lane → falls back to main + fallback := lm.Get("nonexistent") + main := lm.Get(LaneMain) + if fallback != main { + t.Fatal("unknown lane should fall back to main") + } +} + +// --- HasActiveSessionsForAgent --- + +func TestScheduler_HasActiveSessionsForAgent(t *testing.T) { + blockCh := make(chan struct{}) + runFn := func(ctx context.Context, req agent.RunRequest) (*agent.RunResult, error) { + <-blockCh + return &agent.RunResult{}, nil + } + + cfg := DefaultQueueConfig() + cfg.DebounceMs = 0 + sched := NewScheduler(nil, cfg, runFn) + + sched.Schedule(context.Background(), LaneMain, agent.RunRequest{ + SessionKey: "agent:agent-123:scope1", + RunID: "run-1", + }) + time.Sleep(20 * time.Millisecond) // let run start + + if !sched.HasActiveSessionsForAgent("agent-123") { + t.Fatal("expected active sessions for agent-123") + } + if sched.HasActiveSessionsForAgent("other-agent") { + t.Fatal("expected no active sessions for other-agent") + } + + close(blockCh) + time.Sleep(20 * time.Millisecond) + + if sched.HasActiveSessionsForAgent("agent-123") { + t.Fatal("expected no active sessions after completion") + } +} + +// --- Debounce collapsing --- + +func TestSessionQueue_Debounce_CollapsesRapidMessages(t *testing.T) { + var runCount atomic.Int32 + runFn := func(ctx context.Context, req agent.RunRequest) (*agent.RunResult, error) { + runCount.Add(1) + return &agent.RunResult{}, nil + } + + cfg := QueueConfig{ + Mode: QueueModeQueue, + Cap: 10, + Drop: DropOld, + DebounceMs: 200, // 200ms debounce + MaxConcurrent: 1, + } + laneMgr := NewLaneManager([]LaneConfig{{Name: LaneMain, Concurrency: 10}}) + sq := NewSessionQueue("test", LaneMain, cfg, laneMgr, runFn) + + ctx := context.Background() + + // Send 5 rapid messages within debounce window + var channels []<-chan RunOutcome + for i := 0; i < 5; i++ { + ch := sq.Enqueue(ctx, agent.RunRequest{ + RunID: "r" + string(rune('0'+i)), + SessionKey: "test", + }) + channels = append(channels, ch) + } + + // Wait for debounce + execution of all queued messages + for i, ch := range channels { + select { + case <-ch: + case <-time.After(3 * time.Second): + t.Fatalf("message %d timed out", i) + } + } + + // Debounce collapses the initial scheduleNext calls into one timer fire, + // but all 5 messages still execute sequentially. The key behavior is that + // execution doesn't start until after debounce delay (200ms), not immediately. + if got := runCount.Load(); got != 5 { + t.Fatalf("all 5 messages should eventually run, got %d", got) + } +} + +// --- Interrupt mode --- + +func TestSessionQueue_InterruptMode(t *testing.T) { + var runIDs []string + var mu sync.Mutex + + runFn := func(ctx context.Context, req agent.RunRequest) (*agent.RunResult, error) { + mu.Lock() + runIDs = append(runIDs, req.RunID) + mu.Unlock() + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(100 * time.Millisecond): + return &agent.RunResult{}, nil + } + } + + cfg := QueueConfig{ + Mode: QueueModeInterrupt, + Cap: 10, + DebounceMs: 0, + MaxConcurrent: 1, + } + laneMgr := NewLaneManager([]LaneConfig{{Name: LaneMain, Concurrency: 10}}) + sq := NewSessionQueue("test", LaneMain, cfg, laneMgr, runFn) + + ctx := context.Background() + sq.Enqueue(ctx, agent.RunRequest{RunID: "r1", SessionKey: "test"}) + time.Sleep(20 * time.Millisecond) + + // Interrupt: should cancel r1 and start r2 + ch2 := sq.Enqueue(ctx, agent.RunRequest{RunID: "r2", SessionKey: "test"}) + + outcome := <-ch2 + if outcome.Err != nil { + t.Fatalf("r2 should complete successfully: %v", outcome.Err) + } +} diff --git a/internal/sessions/manager_test.go b/internal/sessions/manager_test.go new file mode 100644 index 00000000..e392b836 --- /dev/null +++ b/internal/sessions/manager_test.go @@ -0,0 +1,467 @@ +package sessions + +import ( + "context" + "os" + "sync" + "testing" + + "github.com/nextlevelbuilder/goclaw/internal/providers" +) + +// --- SessionKey --- + +func TestSessionKey(t *testing.T) { + got := SessionKey("agent-123", "telegram:direct:chat456") + want := "agent:agent-123:telegram:direct:chat456" + if got != want { + t.Fatalf("SessionKey = %q, want %q", got, want) + } +} + +// --- GetOrCreate idempotency --- + +func TestGetOrCreate_Idempotent(t *testing.T) { + m := NewManager("") + ctx := context.Background() + + s1 := m.GetOrCreate(ctx, "agent:a1:scope1") + s2 := m.GetOrCreate(ctx, "agent:a1:scope1") + + if s1 != s2 { + t.Fatal("GetOrCreate should return same session pointer for same key") + } +} + +func TestGetOrCreate_DifferentKeys(t *testing.T) { + m := NewManager("") + ctx := context.Background() + + s1 := m.GetOrCreate(ctx, "agent:a1:scope1") + s2 := m.GetOrCreate(ctx, "agent:a2:scope2") + + if s1 == s2 { + t.Fatal("different keys should return different sessions") + } +} + +// --- AddMessage --- + +func TestAddMessage_CreatesSession(t *testing.T) { + m := NewManager("") + ctx := context.Background() + + m.AddMessage(ctx, "agent:a1:s1", providers.Message{Role: "user", Content: "hello"}) + + history := m.GetHistory(ctx, "agent:a1:s1") + if len(history) != 1 { + t.Fatalf("expected 1 message, got %d", len(history)) + } + if history[0].Content != "hello" { + t.Fatalf("content mismatch: got %q", history[0].Content) + } +} + +func TestAddMessage_AppendsToExisting(t *testing.T) { + m := NewManager("") + ctx := context.Background() + + m.AddMessage(ctx, "key1", providers.Message{Role: "user", Content: "msg1"}) + m.AddMessage(ctx, "key1", providers.Message{Role: "assistant", Content: "msg2"}) + m.AddMessage(ctx, "key1", providers.Message{Role: "user", Content: "msg3"}) + + history := m.GetHistory(ctx, "key1") + if len(history) != 3 { + t.Fatalf("expected 3 messages, got %d", len(history)) + } +} + +// --- Concurrent AddMessage: messages must not be lost --- + +func TestAddMessage_ConcurrentSafety(t *testing.T) { + m := NewManager("") + ctx := context.Background() + key := "agent:concurrent:test" + + const n = 100 + var wg sync.WaitGroup + wg.Add(n) + for i := 0; i < n; i++ { + go func(i int) { + defer wg.Done() + m.AddMessage(ctx, key, providers.Message{Role: "user", Content: "msg"}) + }(i) + } + wg.Wait() + + history := m.GetHistory(ctx, key) + if len(history) != n { + t.Fatalf("expected %d messages (no loss), got %d", n, len(history)) + } +} + +// --- GetHistory returns defensive copy --- + +func TestGetHistory_DefensiveCopy(t *testing.T) { + m := NewManager("") + ctx := context.Background() + key := "agent:a1:s1" + + m.AddMessage(ctx, key, providers.Message{Role: "user", Content: "original"}) + + history := m.GetHistory(ctx, key) + history[0].Content = "mutated" // mutate the copy + + // Original should be unchanged + original := m.GetHistory(ctx, key) + if original[0].Content != "original" { + t.Fatal("GetHistory should return a copy, not a reference to internal slice") + } +} + +func TestGetHistory_NonExistentSession(t *testing.T) { + m := NewManager("") + history := m.GetHistory(context.Background(), "nonexistent") + if history != nil { + t.Fatalf("expected nil for non-existent session, got %v", history) + } +} + +// --- SetLabel / SetSummary on missing session: silent no-op --- + +func TestSetLabel_MissingSession_SilentNoOp(t *testing.T) { + m := NewManager("") + // Should not panic + m.SetLabel(context.Background(), "nonexistent", "label") +} + +func TestSetSummary_MissingSession_SilentNoOp(t *testing.T) { + m := NewManager("") + m.SetSummary(context.Background(), "nonexistent", "summary") +} + +// --- Metadata accumulation --- + +func TestAccumulateTokens(t *testing.T) { + m := NewManager("") + ctx := context.Background() + key := "agent:a1:s1" + + m.GetOrCreate(ctx, key) + m.AccumulateTokens(ctx, key, 100, 50) + m.AccumulateTokens(ctx, key, 200, 80) + + m.mu.RLock() + s := m.sessions[key] + m.mu.RUnlock() + + if s.InputTokens != 300 { + t.Fatalf("input tokens: got %d, want 300", s.InputTokens) + } + if s.OutputTokens != 130 { + t.Fatalf("output tokens: got %d, want 130", s.OutputTokens) + } +} + +// --- CompactionCount tracking --- + +func TestCompactionCount(t *testing.T) { + m := NewManager("") + ctx := context.Background() + key := "agent:a1:s1" + + m.GetOrCreate(ctx, key) + + if m.GetCompactionCount(ctx, key) != 0 { + t.Fatal("initial compaction count should be 0") + } + + m.IncrementCompaction(ctx, key) + m.IncrementCompaction(ctx, key) + + if m.GetCompactionCount(ctx, key) != 2 { + t.Fatalf("compaction count: got %d, want 2", m.GetCompactionCount(ctx, key)) + } + + // MemoryFlushCompactionCount tracks separately + if m.GetMemoryFlushCompactionCount(ctx, key) != 0 { + t.Fatal("memory flush compaction should start at 0") + } + + m.SetMemoryFlushDone(ctx, key) + if m.GetMemoryFlushCompactionCount(ctx, key) != 2 { + t.Fatalf("memory flush should match current compaction count: got %d", m.GetMemoryFlushCompactionCount(ctx, key)) + } +} + +func TestGetMemoryFlushCompactionCount_NonExistent(t *testing.T) { + m := NewManager("") + got := m.GetMemoryFlushCompactionCount(context.Background(), "nonexistent") + if got != -1 { + t.Fatalf("expected -1 for non-existent session, got %d", got) + } +} + +// --- TruncateHistory --- + +func TestTruncateHistory(t *testing.T) { + m := NewManager("") + ctx := context.Background() + key := "agent:a1:s1" + + for i := 0; i < 10; i++ { + m.AddMessage(ctx, key, providers.Message{Role: "user", Content: "msg"}) + } + + m.TruncateHistory(ctx, key, 3) + history := m.GetHistory(ctx, key) + if len(history) != 3 { + t.Fatalf("expected 3 messages after truncate, got %d", len(history)) + } +} + +func TestTruncateHistory_KeepZero(t *testing.T) { + m := NewManager("") + ctx := context.Background() + key := "agent:a1:s1" + + m.AddMessage(ctx, key, providers.Message{Role: "user", Content: "msg"}) + m.TruncateHistory(ctx, key, 0) + + history := m.GetHistory(ctx, key) + if len(history) != 0 { + t.Fatalf("expected 0 messages, got %d", len(history)) + } +} + +// --- Reset --- + +func TestReset_ClearsHistoryAndSummary(t *testing.T) { + m := NewManager("") + ctx := context.Background() + key := "agent:a1:s1" + + m.AddMessage(ctx, key, providers.Message{Role: "user", Content: "msg"}) + m.SetSummary(ctx, key, "summary text") + + m.Reset(ctx, key) + + history := m.GetHistory(ctx, key) + if len(history) != 0 { + t.Fatalf("expected empty history after reset, got %d messages", len(history)) + } + if m.GetSummary(ctx, key) != "" { + t.Fatal("expected empty summary after reset") + } +} + +// --- Delete --- + +func TestDelete_RemovesSession(t *testing.T) { + m := NewManager("") + ctx := context.Background() + key := "agent:a1:s1" + + m.GetOrCreate(ctx, key) + if err := m.Delete(ctx, key); err != nil { + t.Fatalf("delete error: %v", err) + } + + history := m.GetHistory(ctx, key) + if history != nil { + t.Fatal("expected nil history after delete") + } +} + +// --- List --- + +func TestList_FiltersByAgentID(t *testing.T) { + m := NewManager("") + ctx := context.Background() + + m.GetOrCreate(ctx, "agent:a1:scope1") + m.GetOrCreate(ctx, "agent:a1:scope2") + m.GetOrCreate(ctx, "agent:a2:scope1") + + all := m.List(ctx, "") + if len(all) != 3 { + t.Fatalf("expected 3 sessions, got %d", len(all)) + } + + filtered := m.List(ctx, "a1") + if len(filtered) != 2 { + t.Fatalf("expected 2 sessions for agent a1, got %d", len(filtered)) + } +} + +// --- ContextWindow --- + +func TestContextWindow_SetAndGet(t *testing.T) { + m := NewManager("") + ctx := context.Background() + key := "agent:a1:s1" + + m.GetOrCreate(ctx, key) + if m.GetContextWindow(ctx, key) != 0 { + t.Fatal("initial context window should be 0") + } + + m.SetContextWindow(ctx, key, 200000) + if m.GetContextWindow(ctx, key) != 200000 { + t.Fatalf("context window mismatch: got %d", m.GetContextWindow(ctx, key)) + } +} + +// --- Persistence: Save and load roundtrip --- + +func TestSave_LoadAll_Roundtrip(t *testing.T) { + dir := t.TempDir() + + // Create manager and save a session + m1 := NewManager(dir) + ctx := context.Background() + key := "agent:a1:scope1" + + m1.AddMessage(ctx, key, providers.Message{Role: "user", Content: "hello"}) + m1.AddMessage(ctx, key, providers.Message{Role: "assistant", Content: "hi"}) + m1.SetLabel(ctx, key, "test-label") + m1.AccumulateTokens(ctx, key, 100, 50) + + if err := m1.Save(ctx, key); err != nil { + t.Fatalf("save error: %v", err) + } + + // Create new manager from same directory → should load the session + m2 := NewManager(dir) + history := m2.GetHistory(ctx, key) + if len(history) != 2 { + t.Fatalf("expected 2 messages after load, got %d", len(history)) + } + if history[0].Content != "hello" || history[1].Content != "hi" { + t.Fatal("message content mismatch after load") + } + + m2.mu.RLock() + s := m2.sessions[key] + m2.mu.RUnlock() + if s.Label != "test-label" { + t.Fatalf("label mismatch: got %q", s.Label) + } + if s.InputTokens != 100 { + t.Fatalf("input tokens mismatch: got %d", s.InputTokens) + } +} + +func TestDelete_RemovesFile(t *testing.T) { + dir := t.TempDir() + m := NewManager(dir) + ctx := context.Background() + key := "agent:a1:scope1" + + m.AddMessage(ctx, key, providers.Message{Role: "user", Content: "msg"}) + _ = m.Save(ctx, key) + + // File should exist + files, _ := os.ReadDir(dir) + if len(files) == 0 { + t.Fatal("expected session file to exist") + } + + _ = m.Delete(ctx, key) + + // File should be gone + files, _ = os.ReadDir(dir) + jsonCount := 0 + for _, f := range files { + if !f.IsDir() { + jsonCount++ + } + } + if jsonCount != 0 { + t.Fatalf("expected no session files after delete, got %d", jsonCount) + } +} + +// --- UpdateMetadata --- + +func TestUpdateMetadata_PartialUpdate(t *testing.T) { + m := NewManager("") + ctx := context.Background() + key := "agent:a1:s1" + + m.GetOrCreate(ctx, key) + m.UpdateMetadata(ctx, key, "claude-3", "anthropic", "telegram") + + m.mu.RLock() + s := m.sessions[key] + m.mu.RUnlock() + + if s.Model != "claude-3" || s.Provider != "anthropic" || s.Channel != "telegram" { + t.Fatalf("metadata mismatch: model=%q provider=%q channel=%q", s.Model, s.Provider, s.Channel) + } + + // Partial update: only model, rest unchanged + m.UpdateMetadata(ctx, key, "gpt-4", "", "") + m.mu.RLock() + s = m.sessions[key] + m.mu.RUnlock() + + if s.Model != "gpt-4" { + t.Fatalf("model should update: got %q", s.Model) + } + if s.Provider != "anthropic" { + t.Fatalf("provider should stay unchanged: got %q", s.Provider) + } +} + +// --- SetSpawnInfo --- + +func TestSetSpawnInfo(t *testing.T) { + m := NewManager("") + ctx := context.Background() + key := "agent:a1:s1" + + m.GetOrCreate(ctx, key) + m.SetSpawnInfo(ctx, key, "parent-run-123", 2) + + m.mu.RLock() + s := m.sessions[key] + m.mu.RUnlock() + + if s.SpawnedBy != "parent-run-123" { + t.Fatalf("spawnedBy: got %q", s.SpawnedBy) + } + if s.SpawnDepth != 2 { + t.Fatalf("spawnDepth: got %d", s.SpawnDepth) + } +} + +// --- LastPromptTokens --- + +func TestLastPromptTokens(t *testing.T) { + m := NewManager("") + ctx := context.Background() + key := "agent:a1:s1" + + m.GetOrCreate(ctx, key) + tokens, msgCount := m.GetLastPromptTokens(ctx, key) + if tokens != 0 || msgCount != 0 { + t.Fatal("initial values should be 0") + } + + m.SetLastPromptTokens(ctx, key, 5000, 42) + tokens, msgCount = m.GetLastPromptTokens(ctx, key) + if tokens != 5000 || msgCount != 42 { + t.Fatalf("got tokens=%d, msgCount=%d", tokens, msgCount) + } +} + +// --- sanitizeFilename --- + +func TestSanitizeFilename(t *testing.T) { + got := sanitizeFilename("agent:a1:telegram:direct:123") + want := "agent_a1_telegram_direct_123" + if got != want { + t.Fatalf("sanitizeFilename = %q, want %q", got, want) + } +}