diff --git a/cmd/gateway.go b/cmd/gateway.go index 7f612e21..da35b29e 100644 --- a/cmd/gateway.go +++ b/cmd/gateway.go @@ -424,6 +424,7 @@ func runGateway() { // Register all RPC methods server.SetLogTee(logTee) + server.SetRuntimeLogsHandler(httpapi.NewRuntimeLogsHandler(logTee)) pairingMethods, heartbeatMethods, chatMethods, cfgPermsMethods := registerAllMethods(server, agentRouter, pgStores.Sessions, pgStores.Cron, pgStores.Pairing, cfg, cfgPath, workspace, dataDir, msgBus, execApprovalMgr, pgStores.Agents, pgStores.Skills, pgStores.ConfigSecrets, pgStores.Teams, contextFileInterceptor, logTee, pgStores.Heartbeats, pgStores.ConfigPermissions, pgStores.SystemConfigs, pgStores.Tenants, pgStores.SkillTenantCfgs, audioMgr) // Phase 3: Agent hooks RPC methods (hooks.list/create/update/delete/toggle/test/history). diff --git a/docs/18-http-api.md b/docs/18-http-api.md index f05157af..333be42d 100644 --- a/docs/18-http-api.md +++ b/docs/18-http-api.md @@ -568,9 +568,32 @@ Read-only session listing is available over HTTP for automation clients. | Method | Path | Description | |--------|------|-------------| | `GET` | `/v1/sessions` | List sessions with `agentId`/`agent_id`, `channel`, `limit`, and `offset` filters | +| `POST` | `/v1/chat/sessions/{key}/branch` | Branch an existing chat session at `up_to_index` | +| `GET` | `/v1/chat/sessions/{key}/history/follow` | Poll session history after an index cursor | Admins and system-level admin API keys can list all sessions within the resolved tenant. Non-admin callers must have an effective `X-GoClaw-User-Id` context and are filtered to their own sessions. +Session branch request: + +```json +{ + "new_session_key": "optional agent:{sameAgentKey}:...", + "up_to_index": 12, + "label": "optional label", + "metadata": {} +} +``` + +If `new_session_key` is omitted, the server generates `agent:{agentKey}:branch:direct:{uuid}`. `up_to_index` copies `messages[0:up_to_index]`; invalid ranges are rejected and existing target keys are not overwritten. + +History follow uses `cursor` as the count of already consumed messages: + +``` +GET /v1/chat/sessions/{key}/history/follow?cursor=12&limit=50 +``` + +If `cursor > total`, the response sets `reset: true` and `next_cursor` to the current total. + --- ## 12. Episodic Memory @@ -978,9 +1001,21 @@ Accepts partial updates. Flag keys are validated against recognized v3 flags. |--------|------|-------------| | `GET` | `/v1/channels/instances/{id}/writers/groups` | List group file writers | | `GET` | `/v1/channels/instances/{id}/writers` | List writers for group | +| `POST` | `/v1/channels/instances/{id}/writers/test` | Test whether a user is a writer for a group | | `POST` | `/v1/channels/instances/{id}/writers` | Add writer to group | | `DELETE` | `/v1/channels/instances/{id}/writers/{userId}` | Remove writer | +Writer test request: + +```json +{ + "group_id": "group:telegram:-100123", + "user_id": "386246614" +} +``` + +Response includes `allowed`, `reason`, `instance_id`, `agent_id`, `group_id`, `user_id`, and `writer_count`. Stable reasons: `writer`, `not_writer`, `no_writers_configured`, `invalid_group`. + **Supported channels:** `telegram`, `discord`, `slack`, `whatsapp`, `zalo_oa`, `zalo_personal`, `feishu` Credentials are masked in HTTP responses. @@ -1215,6 +1250,16 @@ Follow response: | Method | Path | Description | |--------|------|-------------| | `GET` | `/v1/activity` | List activity audit logs (filterable) | +| `GET` | `/v1/activity/aggregate` | Aggregate activity logs by action, actor type, entity type, or admin-only actor ID | +| `GET` | `/v1/logs/runtime/aggregate` | Aggregate recent in-memory runtime logs by level or source | + +Activity aggregate query parameters: +- `group_by`: `action`, `actor_type`, `entity_type`, or `actor_id` (admin only) +- `from`, `to`: optional RFC3339 range, `from` inclusive and `to` exclusive +- `actor_type`, `actor_id`, `action`, `entity_type`, `entity_id`: optional filters; non-admin callers are always scoped to their resolved user ID and must have user context +- `limit`: bucket cap, default 50, max 200 + +Runtime log aggregate is admin-only and ring-buffer based. It returns `retention=ring_buffer`, `capacity`, and `sample_size`; it is not durable log storage. --- diff --git a/docs/project-changelog.md b/docs/project-changelog.md index 3db8a4cb..bd9954e2 100644 --- a/docs/project-changelog.md +++ b/docs/project-changelog.md @@ -6,6 +6,18 @@ Significant changes, features, and fixes in reverse chronological order. ## 2026-05-22 +### CLI P6 backend API unblock + +**Features** + +- Added HTTP session branch and history follow endpoints for automation clients. +- Added read-only channel writer permission testing. +- Added split activity-log and runtime-log aggregate endpoints. + +**Tests** + +- Added focused HTTP, store, and runtime log coverage for branch/follow, writer test, activity aggregate, and LogTee ring aggregation. + ### CI/CD: zuey beta deploy **Features** diff --git a/internal/gateway/log_tee.go b/internal/gateway/log_tee.go index 28228519..4ba33393 100644 --- a/internal/gateway/log_tee.go +++ b/internal/gateway/log_tee.go @@ -3,10 +3,12 @@ package gateway import ( "context" "log/slog" + "sort" "strings" "sync" "time" + runtimelogs "github.com/nextlevelbuilder/goclaw/internal/logs" "github.com/nextlevelbuilder/goclaw/pkg/protocol" ) @@ -25,7 +27,11 @@ var sensitiveKeys = []string{ // while delegating to an underlying handler for normal output. type LogTee struct { inner slog.Handler + state *logTeeState + attrs []slog.Attr +} +type logTeeState struct { mu sync.RWMutex clients map[string]*logSubscriber @@ -46,9 +52,11 @@ type logSubscriber struct { // to any WebSocket clients that have started log tailing. func NewLogTee(inner slog.Handler) *LogTee { return &LogTee{ - inner: inner, - clients: make(map[string]*logSubscriber), - ring: make([]map[string]any, ringBufferSize), + inner: inner, + state: &logTeeState{ + clients: make(map[string]*logSubscriber), + ring: make([]map[string]any, ringBufferSize), + }, } } @@ -58,9 +66,9 @@ func (t *LogTee) Enabled(ctx context.Context, level slog.Level) bool { return true } // Also accept if any subscriber wants this level (e.g. debug). - t.mu.RLock() - defer t.mu.RUnlock() - for _, sub := range t.clients { + t.state.mu.RLock() + defer t.state.mu.RUnlock() + for _, sub := range t.state.clients { if level >= sub.level { return true } @@ -70,35 +78,35 @@ func (t *LogTee) Enabled(ctx context.Context, level slog.Level) bool { func (t *LogTee) Handle(ctx context.Context, r slog.Record) error { // Build entry for WS clients. - t.mu.RLock() - n := len(t.clients) - t.mu.RUnlock() + t.state.mu.RLock() + n := len(t.state.clients) + t.state.mu.RUnlock() needEntry := n > 0 // need to broadcast // Always build entry for ring buffer regardless of subscribers. entry := t.buildEntry(r) // Store in ring buffer. - t.ringMu.Lock() - t.ring[t.ringPos] = entry - t.ringPos = (t.ringPos + 1) % ringBufferSize - if t.ringPos == 0 { - t.ringFul = true + t.state.ringMu.Lock() + t.state.ring[t.state.ringPos] = entry + t.state.ringPos = (t.state.ringPos + 1) % ringBufferSize + if t.state.ringPos == 0 { + t.state.ringFul = true } - t.ringMu.Unlock() + t.state.ringMu.Unlock() // Forward to subscribers. if needEntry { evt := protocol.NewEvent("log", entry) level := r.Level - t.mu.RLock() - for _, sub := range t.clients { + t.state.mu.RLock() + for _, sub := range t.state.clients { if level >= sub.level { sub.client.SendEvent(*evt) } } - t.mu.RUnlock() + t.state.mu.RUnlock() } // Forward to inner handler only if it accepts this level. @@ -117,22 +125,11 @@ func (t *LogTee) buildEntry(r slog.Record) map[string]any { } attrs := map[string]any{} + for _, a := range t.attrs { + applyLogAttr(entry, attrs, a) + } r.Attrs(func(a slog.Attr) bool { - key := a.Key - val := a.Value.String() - - // Extract source hint. - if key == "component" || key == "source" || key == "module" { - entry["source"] = val - return true - } - - // Redact sensitive values. - if isSensitiveKey(key) { - attrs[key] = redactedValue - } else { - attrs[key] = val - } + applyLogAttr(entry, attrs, a) return true }) @@ -144,50 +141,53 @@ func (t *LogTee) buildEntry(r slog.Record) map[string]any { } func (t *LogTee) WithAttrs(attrs []slog.Attr) slog.Handler { + nextAttrs := make([]slog.Attr, 0, len(t.attrs)+len(attrs)) + nextAttrs = append(nextAttrs, t.attrs...) + nextAttrs = append(nextAttrs, attrs...) return &LogTee{ - inner: t.inner.WithAttrs(attrs), - clients: t.clients, - ring: t.ring, + inner: t.inner.WithAttrs(attrs), + state: t.state, + attrs: nextAttrs, } } func (t *LogTee) WithGroup(name string) slog.Handler { return &LogTee{ - inner: t.inner.WithGroup(name), - clients: t.clients, - ring: t.ring, + inner: t.inner.WithGroup(name), + state: t.state, + attrs: append([]slog.Attr(nil), t.attrs...), } } // Subscribe adds a client to the log tailing set at the given level. // Pass slog.LevelInfo for default, slog.LevelDebug for verbose. func (t *LogTee) Subscribe(client *Client, level slog.Level) { - t.mu.Lock() - t.clients[client.ID()] = &logSubscriber{client: client, level: level} - t.mu.Unlock() + t.state.mu.Lock() + t.state.clients[client.ID()] = &logSubscriber{client: client, level: level} + t.state.mu.Unlock() // Replay ring buffer entries at the requested level. - t.ringMu.RLock() + t.state.ringMu.RLock() var entries []map[string]any - if t.ringFul { + if t.state.ringFul { // Buffer is full — read from ringPos (oldest) to ringPos-1 (newest). for i := range ringBufferSize { - idx := (t.ringPos + i) % ringBufferSize - e := t.ring[idx] + idx := (t.state.ringPos + i) % ringBufferSize + e := t.state.ring[idx] if e != nil && logLevelValue(e["level"]) >= level { entries = append(entries, e) } } } else { // Buffer not full — read from 0 to ringPos-1. - for i := 0; i < t.ringPos; i++ { - e := t.ring[i] + for i := 0; i < t.state.ringPos; i++ { + e := t.state.ring[i] if e != nil && logLevelValue(e["level"]) >= level { entries = append(entries, e) } } } - t.ringMu.RUnlock() + t.state.ringMu.RUnlock() for _, e := range entries { client.SendEvent(*protocol.NewEvent("log", e)) @@ -204,9 +204,115 @@ func (t *LogTee) Subscribe(client *Client, level slog.Level) { // Unsubscribe removes a client from the log tailing set. func (t *LogTee) Unsubscribe(clientID string) { - t.mu.Lock() - delete(t.clients, clientID) - t.mu.Unlock() + t.state.mu.Lock() + delete(t.state.clients, clientID) + t.state.mu.Unlock() +} + +func (t *LogTee) AggregateRuntimeLogs(opts runtimelogs.RuntimeAggregateOpts) runtimelogs.RuntimeAggregateResult { + groupBy := opts.GroupBy + if groupBy == "" { + groupBy = "level" + } + entries := t.recentEntries() + counts := map[string]runtimelogs.RuntimeAggregateBucket{} + sampleSize := 0 + for _, e := range entries { + if opts.FromMS > 0 && int64Value(e["timestamp"]) < opts.FromMS { + continue + } + level := stringValue(e["level"]) + if opts.Level != "" && level != opts.Level { + continue + } + source := stringValue(e["source"]) + if opts.Source != "" && source != opts.Source { + continue + } + key := level + if groupBy == "source" { + key = source + if key == "" { + key = "unknown" + } + } + sampleSize++ + b := counts[key] + b.Key = key + b.Count++ + ts := int64Value(e["timestamp"]) + if ts > b.LastSeen { + b.LastSeen = ts + } + counts[key] = b + } + buckets := make([]runtimelogs.RuntimeAggregateBucket, 0, len(counts)) + for _, b := range counts { + buckets = append(buckets, b) + } + sort.Slice(buckets, func(i, j int) bool { + if buckets[i].Count == buckets[j].Count { + return buckets[i].LastSeen > buckets[j].LastSeen + } + return buckets[i].Count > buckets[j].Count + }) + return runtimelogs.RuntimeAggregateResult{ + Source: "runtime", + Retention: "ring_buffer", + Capacity: ringBufferSize, + SampleSize: sampleSize, + GroupBy: groupBy, + Buckets: buckets, + } +} + +func (t *LogTee) recentEntries() []map[string]any { + t.state.ringMu.RLock() + defer t.state.ringMu.RUnlock() + entries := make([]map[string]any, 0, ringBufferSize) + if t.state.ringFul { + for i := range ringBufferSize { + idx := (t.state.ringPos + i) % ringBufferSize + if e := t.state.ring[idx]; e != nil { + entries = append(entries, cloneLogEntry(e)) + } + } + return entries + } + for i := 0; i < t.state.ringPos; i++ { + if e := t.state.ring[i]; e != nil { + entries = append(entries, cloneLogEntry(e)) + } + } + return entries +} + +func cloneLogEntry(in map[string]any) map[string]any { + out := make(map[string]any, len(in)) + for k, v := range in { + out[k] = v + } + return out +} + +func stringValue(v any) string { + s, _ := v.(string) + return s +} + +func int64Value(v any) int64 { + switch n := v.(type) { + case int64: + return n + case int: + return int64(n) + case int32: + return int64(n) + case float64: + return int64(n) + default: + return 0 + } } func levelName(l slog.Level) string { @@ -248,3 +354,17 @@ func isSensitiveKey(key string) bool { } return false } + +func applyLogAttr(entry, attrs map[string]any, a slog.Attr) { + key := a.Key + val := a.Value.String() + if key == "component" || key == "source" || key == "module" { + entry["source"] = val + return + } + if isSensitiveKey(key) { + attrs[key] = redactedValue + return + } + attrs[key] = val +} diff --git a/internal/gateway/log_tee_test.go b/internal/gateway/log_tee_test.go new file mode 100644 index 00000000..7a310b73 --- /dev/null +++ b/internal/gateway/log_tee_test.go @@ -0,0 +1,45 @@ +package gateway + +import ( + "context" + "io" + "log/slog" + "testing" + "time" + + runtimelogs "github.com/nextlevelbuilder/goclaw/internal/logs" +) + +func TestLogTeeAggregateIncludesWithAttrsAndGroupEntries(t *testing.T) { + tee := NewLogTee(slog.NewTextHandler(io.Discard, nil)) + root := slog.New(tee) + root.Info("root", "source", "gateway") + root.With("source", "worker").Warn("child") + root.WithGroup("grouped").Error("grouped", "source", "jobs") + + result := tee.AggregateRuntimeLogs(runtimelogs.RuntimeAggregateOpts{GroupBy: "source"}) + if result.Capacity != 100 || result.Retention != "ring_buffer" { + t.Fatalf("result metadata = %+v", result) + } + got := map[string]int{} + for _, b := range result.Buckets { + got[b.Key] = b.Count + } + if got["gateway"] != 1 || got["worker"] != 1 || got["jobs"] != 1 { + t.Fatalf("buckets = %+v", result.Buckets) + } +} + +func TestLogTeeAggregateRedactsSensitiveAttrs(t *testing.T) { + tee := NewLogTee(slog.NewTextHandler(io.Discard, nil)) + rec := slog.NewRecord(time.Now(), slog.LevelInfo, "secret", 0) + rec.AddAttrs(slog.String("api_token", "leak"), slog.String("source", "test")) + if err := tee.Handle(context.Background(), rec); err != nil { + t.Fatal(err) + } + entries := tee.recentEntries() + attrs, _ := entries[0]["attrs"].(map[string]any) + if attrs["api_token"] != redactedValue { + t.Fatalf("attrs = %+v", attrs) + } +} diff --git a/internal/gateway/server.go b/internal/gateway/server.go index 2289275c..5301e58c 100644 --- a/internal/gateway/server.go +++ b/internal/gateway/server.go @@ -621,6 +621,11 @@ func (s *Server) SetActivityHandler(h *httpapi.ActivityHandler) { s.handlers = append(s.handlers, h) } +// SetRuntimeLogsHandler sets the runtime log aggregate handler. +func (s *Server) SetRuntimeLogsHandler(h *httpapi.RuntimeLogsHandler) { + s.handlers = append(s.handlers, h) +} + // SetSystemConfigsHandler sets the system configs handler. func (s *Server) SetSystemConfigsHandler(h *httpapi.SystemConfigsHandler) { s.handlers = append(s.handlers, h) diff --git a/internal/http/activity.go b/internal/http/activity.go index c21f009e..d716e78c 100644 --- a/internal/http/activity.go +++ b/internal/http/activity.go @@ -3,9 +3,12 @@ package http import ( "net/http" "strconv" + "time" + "github.com/nextlevelbuilder/goclaw/internal/i18n" "github.com/nextlevelbuilder/goclaw/internal/permissions" "github.com/nextlevelbuilder/goclaw/internal/store" + "github.com/nextlevelbuilder/goclaw/pkg/protocol" ) // ActivityHandler handles activity audit log endpoints. @@ -20,6 +23,7 @@ func NewActivityHandler(activity store.ActivityStore) *ActivityHandler { // RegisterRoutes registers activity routes on the given mux. func (h *ActivityHandler) RegisterRoutes(mux *http.ServeMux) { + mux.HandleFunc("GET /v1/activity/aggregate", h.authMiddleware(h.handleAggregate)) mux.HandleFunc("GET /v1/activity", h.authMiddleware(h.handleList)) } @@ -27,6 +31,110 @@ func (h *ActivityHandler) authMiddleware(next http.HandlerFunc) http.HandlerFunc return requireAuth("", next) } +func (h *ActivityHandler) handleAggregate(w http.ResponseWriter, r *http.Request) { + locale := store.LocaleFromContext(r.Context()) + auth := resolveAuth(r) + groupBy := r.URL.Query().Get("group_by") + if !validActivityGroupBy(groupBy) { + writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidRequest, "invalid group_by")) + return + } + if groupBy == "actor_id" && !permissions.HasMinRole(auth.Role, permissions.RoleAdmin) { + writeError(w, http.StatusForbidden, protocol.ErrUnauthorized, i18n.T(locale, i18n.MsgPermissionDenied, "actor_id aggregate")) + return + } + + opts := store.ActivityAggregateOpts{ + GroupBy: groupBy, + ActivityListOpts: store.ActivityListOpts{ + ActorType: r.URL.Query().Get("actor_type"), + ActorID: r.URL.Query().Get("actor_id"), + Action: r.URL.Query().Get("action"), + EntityType: r.URL.Query().Get("entity_type"), + EntityID: r.URL.Query().Get("entity_id"), + Limit: parseActivityLimit(r.URL.Query().Get("limit")), + }, + } + if !permissions.HasMinRole(auth.Role, permissions.RoleAdmin) { + callerID := store.UserIDFromContext(r.Context()) + if callerID == "" { + writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidRequest, "user context required")) + return + } + opts.ActorID = callerID + } + from, ok := parseOptionalRFC3339(r.URL.Query().Get("from")) + if !ok { + writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidRequest, "from must be RFC3339")) + return + } + to, ok := parseOptionalRFC3339(r.URL.Query().Get("to")) + if !ok { + writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidRequest, "to must be RFC3339")) + return + } + if from != nil && to != nil && !from.Before(*to) { + writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidRequest, "from must be before to")) + return + } + opts.From = from + opts.To = to + + buckets, total, err := h.activity.Aggregate(r.Context(), opts) + if err != nil { + writeError(w, http.StatusInternalServerError, protocol.ErrInternal, err.Error()) + return + } + resp := map[string]any{ + "source": "activity", + "group_by": groupBy, + "total": total, + "limit": opts.Limit, + "buckets": buckets, + } + if from != nil { + resp["from"] = from.UTC().Format(time.RFC3339) + } + if to != nil { + resp["to"] = to.UTC().Format(time.RFC3339) + } + writeJSON(w, http.StatusOK, resp) +} + +func validActivityGroupBy(groupBy string) bool { + switch groupBy { + case "action", "actor_type", "entity_type", "actor_id": + return true + default: + return false + } +} + +func parseActivityLimit(raw string) int { + if raw == "" { + return 50 + } + n, err := strconv.Atoi(raw) + if err != nil || n <= 0 { + return 50 + } + if n > 200 { + return 200 + } + return n +} + +func parseOptionalRFC3339(raw string) (*time.Time, bool) { + if raw == "" { + return nil, true + } + t, err := time.Parse(time.RFC3339, raw) + if err != nil { + return nil, false + } + return &t, true +} + func (h *ActivityHandler) handleList(w http.ResponseWriter, r *http.Request) { opts := store.ActivityListOpts{ Limit: 50, diff --git a/internal/http/activity_aggregate_test.go b/internal/http/activity_aggregate_test.go new file mode 100644 index 00000000..3ea303a6 --- /dev/null +++ b/internal/http/activity_aggregate_test.go @@ -0,0 +1,135 @@ +package http + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/nextlevelbuilder/goclaw/internal/crypto" + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +type fakeActivityStore struct { + opts store.ActivityAggregateOpts +} + +func (s *fakeActivityStore) Log(context.Context, *store.ActivityLog) error { return nil } +func (s *fakeActivityStore) List(context.Context, store.ActivityListOpts) ([]store.ActivityLog, error) { + return nil, nil +} +func (s *fakeActivityStore) Count(context.Context, store.ActivityListOpts) (int, error) { + return 0, nil +} +func (s *fakeActivityStore) Aggregate(_ context.Context, opts store.ActivityAggregateOpts) ([]store.ActivityAggregateBucket, int, error) { + s.opts = opts + return []store.ActivityAggregateBucket{{Key: "session.branch", Count: 2, LastSeen: time.Now().UTC()}}, 2, nil +} + +func TestActivityAggregateScopesViewerToCaller(t *testing.T) { + token := "activity-read-key" + setupTestCache(t, map[string]*store.APIKeyData{ + crypto.HashAPIKey(token): {Scopes: []string{"operator.write"}, OwnerID: "caller"}, + }) + activity := &fakeActivityStore{} + mux := http.NewServeMux() + NewActivityHandler(activity).RegisterRoutes(mux) + + req := httptest.NewRequest(http.MethodGet, "/v1/activity/aggregate?group_by=action&actor_id=other&limit=999", nil) + req.Header.Set("Authorization", "Bearer "+token) + rec := httptest.NewRecorder() + mux.ServeHTTP(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200; body=%s", rec.Code, rec.Body.String()) + } + if activity.opts.ActorID != "caller" { + t.Fatalf("ActorID = %q, want caller", activity.opts.ActorID) + } + if activity.opts.Limit != 200 { + t.Fatalf("Limit = %d, want 200", activity.opts.Limit) + } +} + +func TestActivityAggregateRejectsActorIDGroupForViewer(t *testing.T) { + token := "activity-viewer-key" + setupTestCache(t, map[string]*store.APIKeyData{ + crypto.HashAPIKey(token): {Scopes: []string{"operator.write"}, OwnerID: "caller"}, + }) + mux := http.NewServeMux() + NewActivityHandler(&fakeActivityStore{}).RegisterRoutes(mux) + + req := httptest.NewRequest(http.MethodGet, "/v1/activity/aggregate?group_by=actor_id", nil) + req.Header.Set("Authorization", "Bearer "+token) + rec := httptest.NewRecorder() + mux.ServeHTTP(rec, req) + + if rec.Code != http.StatusForbidden { + t.Fatalf("status = %d, want 403; body=%s", rec.Code, rec.Body.String()) + } +} + +func TestActivityAggregateRejectsUnboundNonAdminKey(t *testing.T) { + token := "activity-unbound-key" + setupTestCache(t, map[string]*store.APIKeyData{ + crypto.HashAPIKey(token): {Scopes: []string{"operator.write"}}, + }) + mux := http.NewServeMux() + NewActivityHandler(&fakeActivityStore{}).RegisterRoutes(mux) + + req := httptest.NewRequest(http.MethodGet, "/v1/activity/aggregate?group_by=action", nil) + req.Header.Set("Authorization", "Bearer "+token) + rec := httptest.NewRecorder() + mux.ServeHTTP(rec, req) + + if rec.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want 400; body=%s", rec.Code, rec.Body.String()) + } +} + +func TestActivityAggregateRejectsInvalidRange(t *testing.T) { + token := "activity-admin-key" + setupTestCache(t, map[string]*store.APIKeyData{ + crypto.HashAPIKey(token): {Scopes: []string{"operator.admin"}, OwnerID: "admin"}, + }) + mux := http.NewServeMux() + NewActivityHandler(&fakeActivityStore{}).RegisterRoutes(mux) + + req := httptest.NewRequest(http.MethodGet, "/v1/activity/aggregate?group_by=action&from=2026-05-23T00:00:00Z&to=2026-05-22T00:00:00Z", nil) + req.Header.Set("Authorization", "Bearer "+token) + rec := httptest.NewRecorder() + mux.ServeHTTP(rec, req) + + if rec.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want 400; body=%s", rec.Code, rec.Body.String()) + } +} + +func TestActivityAggregateResponseShape(t *testing.T) { + token := "activity-admin-shape-key" + setupTestCache(t, map[string]*store.APIKeyData{ + crypto.HashAPIKey(token): {Scopes: []string{"operator.admin"}, OwnerID: "admin"}, + }) + mux := http.NewServeMux() + NewActivityHandler(&fakeActivityStore{}).RegisterRoutes(mux) + + req := httptest.NewRequest(http.MethodGet, "/v1/activity/aggregate?group_by=action", nil) + req.Header.Set("Authorization", "Bearer "+token) + rec := httptest.NewRecorder() + mux.ServeHTTP(rec, req) + + var body struct { + Source string `json:"source"` + GroupBy string `json:"group_by"` + Total int `json:"total"` + Buckets []any `json:"buckets"` + } + if err := json.NewDecoder(rec.Body).Decode(&body); err != nil { + t.Fatal(err) + } + if body.Source != "activity" || body.GroupBy != "action" || body.Total != 2 || len(body.Buckets) != 1 { + t.Fatalf("body = %+v", body) + } +} diff --git a/internal/http/api_contracts_test.go b/internal/http/api_contracts_test.go index 863031d2..72a5b0b1 100644 --- a/internal/http/api_contracts_test.go +++ b/internal/http/api_contracts_test.go @@ -195,9 +195,12 @@ func (s *recordingSessionStore) SetLabel(context.Context, string, string) func (s *recordingSessionStore) SetAgentInfo(context.Context, string, uuid.UUID, string) {} func (s *recordingSessionStore) TruncateHistory(context.Context, string, int) {} func (s *recordingSessionStore) SetHistory(context.Context, string, []providers.Message) {} -func (s *recordingSessionStore) Reset(context.Context, string) {} -func (s *recordingSessionStore) Delete(context.Context, string) error { return nil } -func (s *recordingSessionStore) Save(context.Context, string) error { return nil } +func (s *recordingSessionStore) BranchSession(context.Context, string, store.SessionBranchOpts) (*store.SessionData, int, error) { + return nil, 0, store.ErrSessionNotFound +} +func (s *recordingSessionStore) Reset(context.Context, string) {} +func (s *recordingSessionStore) Delete(context.Context, string) error { return nil } +func (s *recordingSessionStore) Save(context.Context, string) error { return nil } func (s *recordingSessionStore) UpdateMetadata(context.Context, string, string, string, string) { } func (s *recordingSessionStore) AccumulateTokens(context.Context, string, int64, int64) {} diff --git a/internal/http/channel_instances.go b/internal/http/channel_instances.go index a0df0150..ac88145a 100644 --- a/internal/http/channel_instances.go +++ b/internal/http/channel_instances.go @@ -105,6 +105,7 @@ func (h *ChannelInstancesHandler) RegisterRoutes(mux *http.ServeMux) { if h.configPermStore != nil { mux.HandleFunc("GET /v1/channels/instances/{id}/writers/groups", h.auth(h.handleWriterGroups)) mux.HandleFunc("GET /v1/channels/instances/{id}/writers", h.auth(h.handleListWriters)) + mux.HandleFunc("POST /v1/channels/instances/{id}/writers/test", h.auth(h.handleTestWriter)) mux.HandleFunc("POST /v1/channels/instances/{id}/writers", h.adminAuth(h.handleAddWriter)) mux.HandleFunc("DELETE /v1/channels/instances/{id}/writers/{userId}", h.adminAuth(h.handleRemoveWriter)) } @@ -381,18 +382,26 @@ func maskInstanceHTTP(inst store.ChannelInstanceData) map[string]any { // resolveAgentID looks up the channel instance and returns its agent_id. func (h *ChannelInstancesHandler) resolveAgentID(w http.ResponseWriter, r *http.Request) (uuid.UUID, bool) { + inst, ok := h.resolveInstance(w, r) + if !ok { + return uuid.Nil, false + } + return inst.AgentID, true +} + +func (h *ChannelInstancesHandler) resolveInstance(w http.ResponseWriter, r *http.Request) (*store.ChannelInstanceData, bool) { locale := store.LocaleFromContext(r.Context()) instID, err := uuid.Parse(r.PathValue("id")) if err != nil { writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidID, "instance")) - return uuid.Nil, false + return nil, false } inst, err := h.store.Get(r.Context(), instID) if err != nil { writeError(w, http.StatusNotFound, protocol.ErrNotFound, i18n.T(locale, i18n.MsgInstanceNotFound)) - return uuid.Nil, false + return nil, false } - return inst.AgentID, true + return inst, true } func (h *ChannelInstancesHandler) handleWriterGroups(w http.ResponseWriter, r *http.Request) { @@ -470,6 +479,68 @@ func (h *ChannelInstancesHandler) handleListWriters(w http.ResponseWriter, r *ht writeJSON(w, http.StatusOK, map[string]any{"writers": writers}) } +func (h *ChannelInstancesHandler) handleTestWriter(w http.ResponseWriter, r *http.Request) { + inst, ok := h.resolveInstance(w, r) + if !ok { + return + } + locale := store.LocaleFromContext(r.Context()) + var body struct { + GroupID string `json:"group_id"` + UserID string `json:"user_id"` + } + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<16)).Decode(&body); err != nil { + writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidJSON)) + return + } + if body.GroupID == "" || body.UserID == "" { + writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgRequired, "group_id and user_id")) + return + } + channelName, _, groupOK := channels.ParseGroupScope(body.GroupID) + if !groupOK || channelName != inst.Name { + writeJSON(w, http.StatusOK, map[string]any{ + "allowed": false, + "reason": "invalid_group", + "instance_id": inst.ID.String(), + "agent_id": inst.AgentID.String(), + "group_id": body.GroupID, + "user_id": body.UserID, + "writer_count": 0, + }) + return + } + + writers, err := h.configPermStore.ListFileWriters(r.Context(), inst.AgentID, body.GroupID) + if err != nil { + slog.Error("channel_instances.test_writer", "error", err) + writeError(w, http.StatusInternalServerError, protocol.ErrInternal, i18n.T(locale, i18n.MsgFailedToList, "writers")) + return + } + allowed := false + for _, p := range writers { + if p.Permission == "allow" && p.UserID == body.UserID { + allowed = true + break + } + } + reason := "not_writer" + if allowed { + reason = "writer" + } else if len(writers) == 0 { + reason = "no_writers_configured" + } + writeJSON(w, http.StatusOK, map[string]any{ + "allowed": allowed, + "reason": reason, + "instance_id": inst.ID.String(), + "agent_id": inst.AgentID.String(), + "group_id": body.GroupID, + "user_id": body.UserID, + "writer_count": len(writers), + }) +} + func (h *ChannelInstancesHandler) handleAddWriter(w http.ResponseWriter, r *http.Request) { agentID, ok := h.resolveAgentID(w, r) if !ok { diff --git a/internal/http/channel_instances_writers_test.go b/internal/http/channel_instances_writers_test.go new file mode 100644 index 00000000..39ccccd0 --- /dev/null +++ b/internal/http/channel_instances_writers_test.go @@ -0,0 +1,102 @@ +package http + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/google/uuid" + + "github.com/nextlevelbuilder/goclaw/internal/crypto" + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +type fakeConfigPermStore struct { + writers []store.ConfigPermission +} + +func (s *fakeConfigPermStore) CheckPermission(context.Context, uuid.UUID, string, string, string) (bool, error) { + return false, nil +} +func (s *fakeConfigPermStore) Grant(context.Context, *store.ConfigPermission) error { return nil } +func (s *fakeConfigPermStore) Revoke(context.Context, uuid.UUID, string, string, string) error { + return nil +} +func (s *fakeConfigPermStore) List(context.Context, uuid.UUID, string, string) ([]store.ConfigPermission, error) { + return nil, nil +} +func (s *fakeConfigPermStore) ListFileWriters(context.Context, uuid.UUID, string) ([]store.ConfigPermission, error) { + return s.writers, nil +} + +func TestChannelWriterTestAllowsConfiguredWriter(t *testing.T) { + token := "writer-read-key" + setupTestCache(t, map[string]*store.APIKeyData{ + crypto.HashAPIKey(token): {Scopes: []string{"operator.write"}, OwnerID: "caller"}, + }) + agentID := uuid.New() + instID := uuid.New() + handler := NewChannelInstancesHandler( + &stubChannelInstanceStore{inst: &store.ChannelInstanceData{BaseModel: store.BaseModel{ID: instID}, Name: "telegram", AgentID: agentID}}, + nil, + &fakeConfigPermStore{writers: []store.ConfigPermission{{UserID: "386246614", Permission: "allow"}}}, + nil, nil, nil, + ) + mux := http.NewServeMux() + handler.RegisterRoutes(mux) + + req := httptest.NewRequest(http.MethodPost, "/v1/channels/instances/"+instID.String()+"/writers/test", strings.NewReader(`{"group_id":"group:telegram:-100123","user_id":"386246614"}`)) + req.Header.Set("Authorization", "Bearer "+token) + rec := httptest.NewRecorder() + mux.ServeHTTP(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200; body=%s", rec.Code, rec.Body.String()) + } + var body struct { + Allowed bool `json:"allowed"` + Reason string `json:"reason"` + WriterCount int `json:"writer_count"` + } + if err := json.NewDecoder(rec.Body).Decode(&body); err != nil { + t.Fatal(err) + } + if !body.Allowed || body.Reason != "writer" || body.WriterCount != 1 { + t.Fatalf("body = %+v", body) + } +} + +func TestChannelWriterTestRejectsWrongGroupScope(t *testing.T) { + token := "writer-invalid-key" + setupTestCache(t, map[string]*store.APIKeyData{ + crypto.HashAPIKey(token): {Scopes: []string{"operator.write"}, OwnerID: "caller"}, + }) + instID := uuid.New() + handler := NewChannelInstancesHandler( + &stubChannelInstanceStore{inst: &store.ChannelInstanceData{BaseModel: store.BaseModel{ID: instID}, Name: "telegram", AgentID: uuid.New()}}, + nil, + &fakeConfigPermStore{writers: []store.ConfigPermission{{UserID: "386246614", Permission: "allow"}}}, + nil, nil, nil, + ) + mux := http.NewServeMux() + handler.RegisterRoutes(mux) + + req := httptest.NewRequest(http.MethodPost, "/v1/channels/instances/"+instID.String()+"/writers/test", strings.NewReader(`{"group_id":"group:discord:123","user_id":"386246614"}`)) + req.Header.Set("Authorization", "Bearer "+token) + rec := httptest.NewRecorder() + mux.ServeHTTP(rec, req) + + var body struct { + Allowed bool `json:"allowed"` + Reason string `json:"reason"` + } + if err := json.NewDecoder(rec.Body).Decode(&body); err != nil { + t.Fatal(err) + } + if body.Allowed || body.Reason != "invalid_group" { + t.Fatalf("body = %+v", body) + } +} diff --git a/internal/http/logs.go b/internal/http/logs.go new file mode 100644 index 00000000..2d9ed123 --- /dev/null +++ b/internal/http/logs.go @@ -0,0 +1,60 @@ +package http + +import ( + "net/http" + "time" + + "github.com/nextlevelbuilder/goclaw/internal/i18n" + runtimelogs "github.com/nextlevelbuilder/goclaw/internal/logs" + "github.com/nextlevelbuilder/goclaw/internal/permissions" + "github.com/nextlevelbuilder/goclaw/internal/store" + "github.com/nextlevelbuilder/goclaw/pkg/protocol" +) + +type RuntimeLogSnapshotter interface { + AggregateRuntimeLogs(opts runtimelogs.RuntimeAggregateOpts) runtimelogs.RuntimeAggregateResult +} + +type RuntimeLogsHandler struct { + snapshotter RuntimeLogSnapshotter +} + +func NewRuntimeLogsHandler(snapshotter RuntimeLogSnapshotter) *RuntimeLogsHandler { + return &RuntimeLogsHandler{snapshotter: snapshotter} +} + +func (h *RuntimeLogsHandler) RegisterRoutes(mux *http.ServeMux) { + mux.HandleFunc("GET /v1/logs/runtime/aggregate", requireAuth(permissions.RoleAdmin, h.handleAggregate)) +} + +func (h *RuntimeLogsHandler) handleAggregate(w http.ResponseWriter, r *http.Request) { + locale := store.LocaleFromContext(r.Context()) + if h.snapshotter == nil { + writeError(w, http.StatusServiceUnavailable, protocol.ErrInternal, i18n.T(locale, i18n.MsgInvalidRequest, "runtime logs unavailable")) + return + } + groupBy := r.URL.Query().Get("group_by") + if groupBy == "" { + groupBy = "level" + } + if groupBy != "level" && groupBy != "source" { + writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidRequest, "invalid group_by")) + return + } + fromMS := int64(0) + if raw := r.URL.Query().Get("from"); raw != "" { + from, err := time.Parse(time.RFC3339, raw) + if err != nil { + writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidRequest, "from must be RFC3339")) + return + } + fromMS = from.UnixMilli() + } + result := h.snapshotter.AggregateRuntimeLogs(runtimelogs.RuntimeAggregateOpts{ + GroupBy: groupBy, + Level: r.URL.Query().Get("level"), + Source: r.URL.Query().Get("source"), + FromMS: fromMS, + }) + writeJSON(w, http.StatusOK, result) +} diff --git a/internal/http/openapi_spec.json b/internal/http/openapi_spec.json index 27cdc50e..4ed39720 100644 --- a/internal/http/openapi_spec.json +++ b/internal/http/openapi_spec.json @@ -106,6 +106,54 @@ } } }, + "/v1/chat/sessions/{key}/branch": { + "post": { + "tags": ["Sessions"], + "summary": "Branch a chat session", + "parameters": [{ "name": "key", "in": "path", "required": true, "schema": { "type": "string" } }], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "type": "object", + "required": ["up_to_index"], + "properties": { + "new_session_key": { "type": "string" }, + "up_to_index": { "type": "integer", "minimum": 0 }, + "label": { "type": "string" }, + "metadata": { "type": "object", "additionalProperties": { "type": "string" } } + } + } + } + } + }, + "responses": { + "201": { "description": "Session branched" }, + "400": { "$ref": "#/components/responses/BadRequest" }, + "403": { "$ref": "#/components/responses/Unauthorized" }, + "404": { "$ref": "#/components/responses/NotFound" }, + "409": { "$ref": "#/components/responses/Conflict" } + } + } + }, + "/v1/chat/sessions/{key}/history/follow": { + "get": { + "tags": ["Sessions"], + "summary": "Poll chat session history by index cursor", + "parameters": [ + { "name": "key", "in": "path", "required": true, "schema": { "type": "string" } }, + { "name": "cursor", "in": "query", "schema": { "type": "integer", "minimum": 0, "default": 0 } }, + { "name": "limit", "in": "query", "schema": { "type": "integer", "default": 50, "maximum": 200 } } + ], + "responses": { + "200": { "description": "History page" }, + "400": { "$ref": "#/components/responses/BadRequest" }, + "403": { "$ref": "#/components/responses/Unauthorized" }, + "404": { "$ref": "#/components/responses/NotFound" } + } + } + }, "/v1/api-keys": { "get": { "tags": ["API Keys"], @@ -791,6 +839,33 @@ "responses": { "200": { "description": "Instance deleted" } } } }, + "/v1/channels/instances/{id}/writers/test": { + "post": { + "tags": ["Channels"], + "summary": "Test channel group writer permission", + "parameters": [{ "name": "id", "in": "path", "required": true, "schema": { "type": "string", "format": "uuid" } }], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "type": "object", + "required": ["group_id", "user_id"], + "properties": { + "group_id": { "type": "string", "example": "group:telegram:-100123" }, + "user_id": { "type": "string", "example": "386246614" } + } + } + } + } + }, + "responses": { + "200": { "description": "Writer permission decision" }, + "400": { "$ref": "#/components/responses/BadRequest" }, + "404": { "$ref": "#/components/responses/NotFound" } + } + } + }, "/v1/contacts": { "get": { "tags": ["Channels"], @@ -890,6 +965,46 @@ "responses": { "200": { "description": "Activity log entries" } } } }, + "/v1/activity/aggregate": { + "get": { + "tags": ["Activity"], + "summary": "Aggregate audit activity logs", + "parameters": [ + { "name": "group_by", "in": "query", "required": true, "schema": { "type": "string", "enum": ["action", "actor_type", "entity_type", "actor_id"] } }, + { "name": "from", "in": "query", "schema": { "type": "string", "format": "date-time" } }, + { "name": "to", "in": "query", "schema": { "type": "string", "format": "date-time" } }, + { "name": "actor_type", "in": "query", "schema": { "type": "string" } }, + { "name": "actor_id", "in": "query", "schema": { "type": "string" } }, + { "name": "action", "in": "query", "schema": { "type": "string" } }, + { "name": "entity_type", "in": "query", "schema": { "type": "string" } }, + { "name": "entity_id", "in": "query", "schema": { "type": "string" } }, + { "name": "limit", "in": "query", "schema": { "type": "integer", "default": 50, "maximum": 200 } } + ], + "responses": { + "200": { "description": "Activity aggregate buckets" }, + "400": { "$ref": "#/components/responses/BadRequest" }, + "403": { "$ref": "#/components/responses/Unauthorized" } + } + } + }, + "/v1/logs/runtime/aggregate": { + "get": { + "tags": ["Activity"], + "summary": "Aggregate recent runtime logs from the in-memory ring buffer", + "parameters": [ + { "name": "group_by", "in": "query", "schema": { "type": "string", "enum": ["level", "source"], "default": "level" } }, + { "name": "level", "in": "query", "schema": { "type": "string", "enum": ["debug", "info", "warn", "error"] } }, + { "name": "source", "in": "query", "schema": { "type": "string" } }, + { "name": "from", "in": "query", "schema": { "type": "string", "format": "date-time" } } + ], + "responses": { + "200": { "description": "Runtime log aggregate buckets" }, + "400": { "$ref": "#/components/responses/BadRequest" }, + "403": { "$ref": "#/components/responses/Unauthorized" }, + "503": { "description": "Runtime log tee unavailable" } + } + } + }, "/v1/delegations": { "get": { "tags": ["Activity"], diff --git a/internal/http/sessions.go b/internal/http/sessions.go index 17b6b83b..ee20d1f2 100644 --- a/internal/http/sessions.go +++ b/internal/http/sessions.go @@ -1,12 +1,19 @@ package http import ( + "encoding/json" + "errors" "net/http" "slices" "strconv" + "time" + + "github.com/google/uuid" "github.com/nextlevelbuilder/goclaw/internal/i18n" "github.com/nextlevelbuilder/goclaw/internal/permissions" + "github.com/nextlevelbuilder/goclaw/internal/providers" + "github.com/nextlevelbuilder/goclaw/internal/sessions" "github.com/nextlevelbuilder/goclaw/internal/store" "github.com/nextlevelbuilder/goclaw/pkg/protocol" ) @@ -23,6 +30,8 @@ func NewSessionsHandler(s store.SessionStore, ownerIDs []string) *SessionsHandle func (h *SessionsHandler) RegisterRoutes(mux *http.ServeMux) { mux.HandleFunc("GET /v1/sessions", requireAuth("", h.handleList)) + mux.HandleFunc("POST /v1/chat/sessions/{key}/branch", requireAuth("", h.handleBranch)) + mux.HandleFunc("GET /v1/chat/sessions/{key}/history/follow", requireAuth("", h.handleHistoryFollow)) } func (h *SessionsHandler) handleList(w http.ResponseWriter, r *http.Request) { @@ -57,6 +66,149 @@ func (h *SessionsHandler) handleList(w http.ResponseWriter, r *http.Request) { }) } +func (h *SessionsHandler) handleBranch(w http.ResponseWriter, r *http.Request) { + locale := store.LocaleFromContext(r.Context()) + sourceKey := r.PathValue("key") + sourceAgent, _ := sessions.ParseSessionKey(sourceKey) + if sourceAgent == "" { + writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidRequest, "invalid session key")) + return + } + source := h.sessions.Get(r.Context(), sourceKey) + if source == nil { + writeError(w, http.StatusNotFound, protocol.ErrNotFound, i18n.T(locale, i18n.MsgNotFound, "session", sourceKey)) + return + } + if !h.canAccessSession(r, source) { + writeError(w, http.StatusForbidden, protocol.ErrUnauthorized, i18n.T(locale, i18n.MsgPermissionDenied, "session")) + return + } + + var body struct { + NewSessionKey string `json:"new_session_key"` + UpToIndex *int `json:"up_to_index"` + Label string `json:"label"` + Metadata map[string]string `json:"metadata"` + } + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20)).Decode(&body); err != nil { + writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidJSON)) + return + } + if body.UpToIndex == nil { + writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgRequired, "up_to_index")) + return + } + newKey := body.NewSessionKey + if newKey == "" { + newKey = "agent:" + sourceAgent + ":branch:direct:" + uuid.NewString() + } else if targetAgent, _ := sessions.ParseSessionKey(newKey); targetAgent == "" || targetAgent != sourceAgent { + writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidRequest, "new_session_key must use the same agent key")) + return + } + + brancher, ok := h.sessions.(store.SessionBranchStore) + if !ok { + writeError(w, http.StatusServiceUnavailable, protocol.ErrInternal, i18n.T(locale, i18n.MsgInvalidRequest, "session branching unavailable")) + return + } + branch, copied, err := brancher.BranchSession(r.Context(), sourceKey, store.SessionBranchOpts{ + NewKey: newKey, + UpToIndex: *body.UpToIndex, + Label: body.Label, + Metadata: body.Metadata, + }) + if err != nil { + switch { + case errors.Is(err, store.ErrSessionAlreadyExists): + writeError(w, http.StatusConflict, protocol.ErrFailedPrecondition, i18n.T(locale, i18n.MsgAlreadyExists, "session", newKey)) + case errors.Is(err, store.ErrInvalidSessionBranch): + writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidRequest, "up_to_index out of range")) + case errors.Is(err, store.ErrSessionNotFound): + writeError(w, http.StatusNotFound, protocol.ErrNotFound, i18n.T(locale, i18n.MsgNotFound, "session", sourceKey)) + default: + writeError(w, http.StatusInternalServerError, protocol.ErrInternal, err.Error()) + } + return + } + + writeJSON(w, http.StatusCreated, map[string]any{ + "ok": true, + "source_key": sourceKey, + "session_key": branch.Key, + "copied_messages": copied, + "total_messages": len(source.Messages), + "label": branch.Label, + }) +} + +func (h *SessionsHandler) handleHistoryFollow(w http.ResponseWriter, r *http.Request) { + locale := store.LocaleFromContext(r.Context()) + sessionKey := r.PathValue("key") + sess := h.sessions.Get(r.Context(), sessionKey) + if sess == nil { + writeError(w, http.StatusNotFound, protocol.ErrNotFound, i18n.T(locale, i18n.MsgNotFound, "session", sessionKey)) + return + } + if !h.canAccessSession(r, sess) { + writeError(w, http.StatusForbidden, protocol.ErrUnauthorized, i18n.T(locale, i18n.MsgPermissionDenied, "session")) + return + } + + cursor, ok := parseStrictNonNegativeInt(r.URL.Query().Get("cursor"), 0) + if !ok { + writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidRequest, "cursor must be a non-negative integer")) + return + } + limit, ok := parseStrictPositiveInt(r.URL.Query().Get("limit"), 50, 200) + if !ok { + writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidRequest, "limit must be a positive integer")) + return + } + history := h.sessions.GetHistory(r.Context(), sessionKey) + total := len(history) + if cursor > total { + writeJSON(w, http.StatusOK, map[string]any{ + "session_key": sessionKey, + "cursor": cursor, + "next_cursor": total, + "total": total, + "messages": []any{}, + "reset": true, + "updated": sess.Updated.UTC().Format(time.RFC3339Nano), + }) + return + } + + end := cursor + limit + if end > total { + end = total + } + messages := append([]providers.Message(nil), history[cursor:end]...) + secret := FileSigningKey() + for i := range messages { + messages[i].Content = SignFileURLs(messages[i].Content, secret) + for j := range messages[i].MediaRefs { + messages[i].MediaRefs[j].Path = SignMediaPath(messages[i].MediaRefs[j].Path, secret) + } + } + + writeJSON(w, http.StatusOK, map[string]any{ + "session_key": sessionKey, + "cursor": cursor, + "next_cursor": end, + "total": total, + "messages": messages, + "reset": false, + "updated": sess.Updated.UTC().Format(time.RFC3339Nano), + }) +} + +func (h *SessionsHandler) canAccessSession(r *http.Request, sess *store.SessionData) bool { + userID := store.UserIDFromContext(r.Context()) + role := permissions.Role(store.RoleFromContext(r.Context())) + return canSeeAllHTTP(role, h.ownerIDs, userID) || (userID != "" && sess.UserID == userID) +} + func canSeeAllHTTP(role permissions.Role, ownerIDs []string, userID string) bool { if permissions.HasMinRole(role, permissions.RoleAdmin) { return true @@ -72,6 +224,31 @@ func parsePositiveInt(raw string, fallback int) int { return n } +func parseStrictPositiveInt(raw string, fallback, maxValue int) (int, bool) { + if raw == "" { + return fallback, true + } + n, err := strconv.Atoi(raw) + if err != nil || n <= 0 { + return 0, false + } + if n > maxValue { + n = maxValue + } + return n, true +} + +func parseStrictNonNegativeInt(raw string, fallback int) (int, bool) { + if raw == "" { + return fallback, true + } + n, err := strconv.Atoi(raw) + if err != nil || n < 0 { + return 0, false + } + return n, true +} + func parseNonNegativeInt(raw string, fallback int) int { n, err := strconv.Atoi(raw) if err != nil || n < 0 { diff --git a/internal/http/sessions_branch_follow_test.go b/internal/http/sessions_branch_follow_test.go new file mode 100644 index 00000000..a8199dbe --- /dev/null +++ b/internal/http/sessions_branch_follow_test.go @@ -0,0 +1,138 @@ +package http + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/nextlevelbuilder/goclaw/internal/crypto" + "github.com/nextlevelbuilder/goclaw/internal/providers" + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +type branchSessionStore struct { + recordingSessionStore + data map[string]*store.SessionData +} + +func (s *branchSessionStore) Get(_ context.Context, key string) *store.SessionData { + return s.data[key] +} + +func (s *branchSessionStore) GetHistory(_ context.Context, key string) []providers.Message { + if sess := s.data[key]; sess != nil { + return append([]providers.Message(nil), sess.Messages...) + } + return nil +} + +func (s *branchSessionStore) BranchSession(_ context.Context, sourceKey string, opts store.SessionBranchOpts) (*store.SessionData, int, error) { + source := s.data[sourceKey] + if source == nil { + return nil, 0, store.ErrSessionNotFound + } + if _, exists := s.data[opts.NewKey]; exists { + return nil, 0, store.ErrSessionAlreadyExists + } + if opts.UpToIndex < 0 || opts.UpToIndex > len(source.Messages) { + return nil, 0, store.ErrInvalidSessionBranch + } + msgs := append([]providers.Message(nil), source.Messages[:opts.UpToIndex]...) + branch := &store.SessionData{ + Key: opts.NewKey, + Messages: msgs, + UserID: source.UserID, + Channel: "branch", + Label: opts.Label, + Updated: time.Now().UTC(), + } + s.data[opts.NewKey] = branch + return branch, len(msgs), nil +} + +func TestSessionsBranchCreatesGeneratedBranchAtIndex(t *testing.T) { + token := "session-read-key" + setupTestCache(t, map[string]*store.APIKeyData{ + crypto.HashAPIKey(token): {Scopes: []string{"operator.write"}, OwnerID: "caller"}, + }) + sessions := &branchSessionStore{data: map[string]*store.SessionData{ + "agent:default:ws:direct:abc": { + Key: "agent:default:ws:direct:abc", + UserID: "caller", + Updated: time.Now().UTC(), + Messages: []providers.Message{ + {Role: "user", Content: "one"}, + {Role: "assistant", Content: "two"}, + {Role: "user", Content: "three"}, + }, + }, + }} + mux := http.NewServeMux() + NewSessionsHandler(sessions, nil).RegisterRoutes(mux) + + req := httptest.NewRequest(http.MethodPost, "/v1/chat/sessions/agent:default:ws:direct:abc/branch", strings.NewReader(`{"up_to_index":2,"label":"branch"}`)) + req.Header.Set("Authorization", "Bearer "+token) + rec := httptest.NewRecorder() + mux.ServeHTTP(rec, req) + + if rec.Code != http.StatusCreated { + t.Fatalf("status = %d, want 201; body=%s", rec.Code, rec.Body.String()) + } + var body struct { + SessionKey string `json:"session_key"` + CopiedMessages int `json:"copied_messages"` + TotalMessages int `json:"total_messages"` + } + if err := json.NewDecoder(rec.Body).Decode(&body); err != nil { + t.Fatal(err) + } + if !strings.HasPrefix(body.SessionKey, "agent:default:branch:direct:") { + t.Fatalf("session_key = %q", body.SessionKey) + } + if body.CopiedMessages != 2 || body.TotalMessages != 3 { + t.Fatalf("body = %+v", body) + } +} + +func TestSessionsHistoryFollowCursorReset(t *testing.T) { + token := "session-follow-key" + setupTestCache(t, map[string]*store.APIKeyData{ + crypto.HashAPIKey(token): {Scopes: []string{"operator.write"}, OwnerID: "caller"}, + }) + sessions := &branchSessionStore{data: map[string]*store.SessionData{ + "agent:default:ws:direct:abc": { + Key: "agent:default:ws:direct:abc", + UserID: "caller", + Updated: time.Now().UTC(), + Messages: []providers.Message{ + {Role: "user", Content: "one"}, + }, + }, + }} + mux := http.NewServeMux() + NewSessionsHandler(sessions, nil).RegisterRoutes(mux) + + req := httptest.NewRequest(http.MethodGet, "/v1/chat/sessions/agent:default:ws:direct:abc/history/follow?cursor=9", nil) + req.Header.Set("Authorization", "Bearer "+token) + rec := httptest.NewRecorder() + mux.ServeHTTP(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200; body=%s", rec.Code, rec.Body.String()) + } + var body struct { + Reset bool `json:"reset"` + NextCursor int `json:"next_cursor"` + Total int `json:"total"` + } + if err := json.NewDecoder(rec.Body).Decode(&body); err != nil { + t.Fatal(err) + } + if !body.Reset || body.NextCursor != 1 || body.Total != 1 { + t.Fatalf("body = %+v", body) + } +} diff --git a/internal/logs/runtime.go b/internal/logs/runtime.go new file mode 100644 index 00000000..db3f34de --- /dev/null +++ b/internal/logs/runtime.go @@ -0,0 +1,26 @@ +package logs + +// RuntimeAggregateOpts configures aggregation over the in-memory runtime log ring. +type RuntimeAggregateOpts struct { + GroupBy string + Level string + Source string + FromMS int64 +} + +// RuntimeAggregateBucket is a grouped runtime log count. +type RuntimeAggregateBucket struct { + Key string `json:"key"` + Count int `json:"count"` + LastSeen int64 `json:"last_seen"` +} + +// RuntimeAggregateResult describes the bounded runtime log aggregate. +type RuntimeAggregateResult struct { + Source string `json:"source"` + Retention string `json:"retention"` + Capacity int `json:"capacity"` + SampleSize int `json:"sample_size"` + GroupBy string `json:"group_by"` + Buckets []RuntimeAggregateBucket `json:"buckets"` +} diff --git a/internal/store/activity_store.go b/internal/store/activity_store.go index b203b1e9..8f042c5c 100644 --- a/internal/store/activity_store.go +++ b/internal/store/activity_store.go @@ -28,13 +28,29 @@ type ActivityListOpts struct { Action string EntityType string EntityID string + From *time.Time + To *time.Time Limit int Offset int } +// ActivityAggregateOpts configures audit log aggregation. +type ActivityAggregateOpts struct { + ActivityListOpts + GroupBy string +} + +// ActivityAggregateBucket is a grouped audit log count. +type ActivityAggregateBucket struct { + Key string `json:"key"` + Count int `json:"count"` + LastSeen time.Time `json:"last_seen"` +} + // ActivityStore manages activity audit logs. type ActivityStore interface { Log(ctx context.Context, entry *ActivityLog) error List(ctx context.Context, opts ActivityListOpts) ([]ActivityLog, error) Count(ctx context.Context, opts ActivityListOpts) (int, error) + Aggregate(ctx context.Context, opts ActivityAggregateOpts) ([]ActivityAggregateBucket, int, error) } diff --git a/internal/store/pg/activity_store.go b/internal/store/pg/activity_store.go index 05d9e584..9a640416 100644 --- a/internal/store/pg/activity_store.go +++ b/internal/store/pg/activity_store.go @@ -75,6 +75,46 @@ func (s *PGActivityStore) Count(ctx context.Context, opts store.ActivityListOpts return count, err } +func (s *PGActivityStore) Aggregate(ctx context.Context, opts store.ActivityAggregateOpts) ([]store.ActivityAggregateBucket, int, error) { + column, ok := activityAggregateColumn(opts.GroupBy) + if !ok { + return nil, 0, fmt.Errorf("invalid group_by") + } + limit := opts.Limit + if limit <= 0 { + limit = 50 + } + where, args := buildActivityWhere(ctx, opts.ActivityListOpts) + total, err := s.Count(ctx, opts.ActivityListOpts) + if err != nil { + return nil, 0, err + } + args = append(args, limit) + query := fmt.Sprintf( + `SELECT COALESCE(%s, ''), COUNT(*), MAX(created_at) + FROM activity_logs %s + GROUP BY %s + ORDER BY COUNT(*) DESC, MAX(created_at) DESC + LIMIT $%d`, + column, where, column, len(args), + ) + rows, err := s.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, 0, err + } + defer rows.Close() + + var buckets []store.ActivityAggregateBucket + for rows.Next() { + var b store.ActivityAggregateBucket + if err := rows.Scan(&b.Key, &b.Count, &b.LastSeen); err != nil { + return nil, 0, err + } + buckets = append(buckets, b) + } + return buckets, total, rows.Err() +} + func buildActivityWhere(ctx context.Context, opts store.ActivityListOpts) (string, []any) { var conditions []string var args []any @@ -114,9 +154,34 @@ func buildActivityWhere(ctx context.Context, opts store.ActivityListOpts) (strin args = append(args, opts.EntityID) idx++ } + if opts.From != nil { + conditions = append(conditions, fmt.Sprintf("created_at >= $%d", idx)) + args = append(args, *opts.From) + idx++ + } + if opts.To != nil { + conditions = append(conditions, fmt.Sprintf("created_at < $%d", idx)) + args = append(args, *opts.To) + idx++ + } if len(conditions) == 0 { return "", nil } return "WHERE " + strings.Join(conditions, " AND "), args } + +func activityAggregateColumn(groupBy string) (string, bool) { + switch groupBy { + case "action": + return "action", true + case "actor_type": + return "actor_type", true + case "entity_type": + return "entity_type", true + case "actor_id": + return "actor_id", true + default: + return "", false + } +} diff --git a/internal/store/pg/sessions_branch.go b/internal/store/pg/sessions_branch.go new file mode 100644 index 00000000..a3433aca --- /dev/null +++ b/internal/store/pg/sessions_branch.go @@ -0,0 +1,104 @@ +package pg + +import ( + "context" + "encoding/json" + "maps" + "strconv" + "time" + + "github.com/google/uuid" + + "github.com/nextlevelbuilder/goclaw/internal/providers" + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +func (s *PGSessionStore) BranchSession(ctx context.Context, sourceKey string, opts store.SessionBranchOpts) (*store.SessionData, int, error) { + if opts.NewKey == "" || opts.UpToIndex < 0 { + return nil, 0, store.ErrInvalidSessionBranch + } + if s.Get(ctx, opts.NewKey) != nil { + return nil, 0, store.ErrSessionAlreadyExists + } + + source := s.Get(ctx, sourceKey) + if source == nil { + return nil, 0, store.ErrSessionNotFound + } + + s.mu.RLock() + sourceSnapshot := *source + sourceMsgs := make([]providers.Message, len(source.Messages)) + copy(sourceMsgs, source.Messages) + sourceSnapshot.Messages = sourceMsgs + sourceMeta := make(map[string]string, len(source.Metadata)) + maps.Copy(sourceMeta, source.Metadata) + sourceSnapshot.Metadata = sourceMeta + s.mu.RUnlock() + + if opts.UpToIndex > len(sourceSnapshot.Messages) { + return nil, 0, store.ErrInvalidSessionBranch + } + + now := time.Now().UTC() + copied := make([]providers.Message, opts.UpToIndex) + copy(copied, sourceSnapshot.Messages[:opts.UpToIndex]) + meta := make(map[string]string, len(sourceSnapshot.Metadata)+len(opts.Metadata)+3) + maps.Copy(meta, sourceSnapshot.Metadata) + maps.Copy(meta, opts.Metadata) + meta["branched_from"] = sourceKey + meta["branched_from_index"] = strconv.Itoa(opts.UpToIndex) + meta["branched_at"] = now.Format(time.RFC3339Nano) + + branch := &store.SessionData{ + Key: opts.NewKey, + Messages: copied, + Summary: sourceSnapshot.Summary, + Created: now, + Updated: now, + AgentUUID: sourceSnapshot.AgentUUID, + UserID: sourceSnapshot.UserID, + Model: sourceSnapshot.Model, + Provider: sourceSnapshot.Provider, + Channel: "branch", + Label: opts.Label, + Metadata: meta, + ContextWindow: sourceSnapshot.ContextWindow, + } + if sourceSnapshot.TeamID != nil { + teamID := *sourceSnapshot.TeamID + branch.TeamID = &teamID + } + msgsJSON, _ := json.Marshal(branch.Messages) + metaJSON := []byte("{}") + if len(branch.Metadata) > 0 { + metaJSON, _ = json.Marshal(branch.Metadata) + } + + res, err := s.db.ExecContext(ctx, + `INSERT INTO sessions (id, session_key, messages, summary, model, provider, channel, + input_tokens, output_tokens, compaction_count, + memory_flush_compaction_count, memory_flush_at, + label, spawned_by, spawn_depth, agent_id, user_id, metadata, updated_at, team_id, tenant_id, created_at) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22) + ON CONFLICT (tenant_id, session_key) DO NOTHING`, + uuid.Must(uuid.NewV7()), branch.Key, msgsJSON, + nilStr(branch.Summary), nilStr(branch.Model), nilStr(branch.Provider), nilStr(branch.Channel), + branch.InputTokens, branch.OutputTokens, branch.CompactionCount, + branch.MemoryFlushCompactionCount, branch.MemoryFlushAt, + nilStr(branch.Label), nilStr(branch.SpawnedBy), branch.SpawnDepth, + nilSessionUUID(branch.AgentUUID), nilStr(branch.UserID), metaJSON, branch.Updated, + branch.TeamID, tenantIDForInsert(ctx), branch.Created, + ) + if err != nil { + return nil, 0, err + } + if n, _ := res.RowsAffected(); n == 0 { + return nil, 0, store.ErrSessionAlreadyExists + } + + s.mu.Lock() + s.cache[sessionCacheKey(ctx, opts.NewKey)] = branch + s.mu.Unlock() + return branch, len(copied), nil +} diff --git a/internal/store/session_store.go b/internal/store/session_store.go index c110ceb7..93e95f90 100644 --- a/internal/store/session_store.go +++ b/internal/store/session_store.go @@ -2,12 +2,27 @@ package store import ( "context" + "errors" "time" "github.com/google/uuid" "github.com/nextlevelbuilder/goclaw/internal/providers" ) +var ( + ErrSessionNotFound = errors.New("session not found") + ErrSessionAlreadyExists = errors.New("session already exists") + ErrInvalidSessionBranch = errors.New("invalid session branch") +) + +// SessionBranchOpts configures a durable session branch operation. +type SessionBranchOpts struct { + NewKey string + UpToIndex int + Label string + Metadata map[string]string +} + // SessionData holds conversation state for one session. type SessionData struct { Key string `json:"key" db:"key"` @@ -20,21 +35,21 @@ type SessionData struct { UserID string `json:"userID,omitempty" db:"user_id"` // External user ID (e.g. Telegram user ID) TeamID *uuid.UUID `json:"teamID,omitempty" db:"team_id"` // Team UUID (set for team sessions) - Model string `json:"model,omitempty" db:"model"` - Provider string `json:"provider,omitempty" db:"provider"` - Channel string `json:"channel,omitempty" db:"channel"` - InputTokens int64 `json:"inputTokens,omitempty" db:"input_tokens"` - OutputTokens int64 `json:"outputTokens,omitempty" db:"output_tokens"` - CompactionCount int `json:"compactionCount,omitempty" db:"compaction_count"` - MemoryFlushCompactionCount int `json:"memoryFlushCompactionCount,omitempty" db:"memory_flush_compaction_count"` - MemoryFlushAt int64 `json:"memoryFlushAt,omitempty" db:"-"` - Label string `json:"label,omitempty" db:"label"` + Model string `json:"model,omitempty" db:"model"` + Provider string `json:"provider,omitempty" db:"provider"` + Channel string `json:"channel,omitempty" db:"channel"` + InputTokens int64 `json:"inputTokens,omitempty" db:"input_tokens"` + OutputTokens int64 `json:"outputTokens,omitempty" db:"output_tokens"` + CompactionCount int `json:"compactionCount,omitempty" db:"compaction_count"` + MemoryFlushCompactionCount int `json:"memoryFlushCompactionCount,omitempty" db:"memory_flush_compaction_count"` + MemoryFlushAt int64 `json:"memoryFlushAt,omitempty" db:"-"` + Label string `json:"label,omitempty" db:"label"` SpawnedBy string `json:"spawnedBy,omitempty" db:"spawned_by"` SpawnDepth int `json:"spawnDepth,omitempty" db:"spawn_depth"` Metadata map[string]string `json:"metadata,omitempty" db:"metadata"` // Adaptive throttle: cached per-session so scheduler reads without DB lookup. - ContextWindow int `json:"contextWindow,omitempty" db:"context_window"` // agent's context window (set on first run) + ContextWindow int `json:"contextWindow,omitempty" db:"context_window"` // agent's context window (set on first run) LastPromptTokens int `json:"lastPromptTokens,omitempty" db:"last_prompt_tokens"` // actual prompt tokens from last LLM response LastMessageCount int `json:"lastMessageCount,omitempty" db:"last_message_count"` // message count at time of last LLM call } @@ -75,8 +90,8 @@ type SessionInfoRich struct { InputTokens int64 `json:"inputTokens,omitempty" db:"input_tokens"` OutputTokens int64 `json:"outputTokens,omitempty" db:"output_tokens"` AgentName string `json:"agentName,omitempty" db:"agent_name"` - EstimatedTokens int `json:"estimatedTokens,omitempty" db:"-"` // estimated current context tokens (messages bytes/4 + 12k system prompt) - ContextWindow int `json:"contextWindow,omitempty" db:"context_window"` // agent's context window size + EstimatedTokens int `json:"estimatedTokens,omitempty" db:"-"` // estimated current context tokens (messages bytes/4 + 12k system prompt) + ContextWindow int `json:"contextWindow,omitempty" db:"context_window"` // agent's context window size CompactionCount int `json:"compactionCount,omitempty" db:"compaction_count"` // number of compactions performed } @@ -105,6 +120,12 @@ type SessionCoreStore interface { Save(ctx context.Context, key string) error } +// SessionBranchStore is implemented by durable stores that can clone a session +// into a new branch key without overwriting an existing target. +type SessionBranchStore interface { + BranchSession(ctx context.Context, sourceKey string, opts SessionBranchOpts) (*SessionData, int, error) +} + // SessionMetadataStore manages session metadata, token tracking, and calibration. type SessionMetadataStore interface { UpdateMetadata(ctx context.Context, key, model, provider, channel string) diff --git a/internal/store/sqlitestore/activity.go b/internal/store/sqlitestore/activity.go index cf4b0b9e..04ee0119 100644 --- a/internal/store/sqlitestore/activity.go +++ b/internal/store/sqlitestore/activity.go @@ -80,6 +80,48 @@ func (s *SQLiteActivityStore) Count(ctx context.Context, opts store.ActivityList return count, err } +func (s *SQLiteActivityStore) Aggregate(ctx context.Context, opts store.ActivityAggregateOpts) ([]store.ActivityAggregateBucket, int, error) { + column, ok := activityAggregateColumn(opts.GroupBy) + if !ok { + return nil, 0, fmt.Errorf("invalid group_by") + } + limit := opts.Limit + if limit <= 0 { + limit = 50 + } + where, args := buildActivityWhere(ctx, opts.ActivityListOpts) + total, err := s.Count(ctx, opts.ActivityListOpts) + if err != nil { + return nil, 0, err + } + args = append(args, limit) + query := fmt.Sprintf( + `SELECT COALESCE(%s, ''), COUNT(*), MAX(created_at) + FROM activity_logs %s + GROUP BY %s + ORDER BY COUNT(*) DESC, MAX(created_at) DESC + LIMIT ?`, + column, where, column, + ) + rows, err := s.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, 0, err + } + defer rows.Close() + + var buckets []store.ActivityAggregateBucket + for rows.Next() { + var b store.ActivityAggregateBucket + var lastSeen sqliteTime + if err := rows.Scan(&b.Key, &b.Count, &lastSeen); err != nil { + return nil, 0, err + } + b.LastSeen = lastSeen.Time + buckets = append(buckets, b) + } + return buckets, total, rows.Err() +} + func buildActivityWhere(ctx context.Context, opts store.ActivityListOpts) (string, []any) { var conditions []string var args []any @@ -112,9 +154,32 @@ func buildActivityWhere(ctx context.Context, opts store.ActivityListOpts) (strin conditions = append(conditions, "entity_id = ?") args = append(args, opts.EntityID) } + if opts.From != nil { + conditions = append(conditions, "created_at >= ?") + args = append(args, *opts.From) + } + if opts.To != nil { + conditions = append(conditions, "created_at < ?") + args = append(args, *opts.To) + } if len(conditions) == 0 { return "", nil } return "WHERE " + strings.Join(conditions, " AND "), args } + +func activityAggregateColumn(groupBy string) (string, bool) { + switch groupBy { + case "action": + return "action", true + case "actor_type": + return "actor_type", true + case "entity_type": + return "entity_type", true + case "actor_id": + return "actor_id", true + default: + return "", false + } +} diff --git a/internal/store/sqlitestore/sessions_branch.go b/internal/store/sqlitestore/sessions_branch.go new file mode 100644 index 00000000..d810e2a7 --- /dev/null +++ b/internal/store/sqlitestore/sessions_branch.go @@ -0,0 +1,107 @@ +//go:build sqlite || sqliteonly + +package sqlitestore + +import ( + "context" + "encoding/json" + "maps" + "strconv" + "time" + + "github.com/google/uuid" + + "github.com/nextlevelbuilder/goclaw/internal/providers" + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +func (s *SQLiteSessionStore) BranchSession(ctx context.Context, sourceKey string, opts store.SessionBranchOpts) (*store.SessionData, int, error) { + if opts.NewKey == "" || opts.UpToIndex < 0 { + return nil, 0, store.ErrInvalidSessionBranch + } + if s.Get(ctx, opts.NewKey) != nil { + return nil, 0, store.ErrSessionAlreadyExists + } + + source := s.Get(ctx, sourceKey) + if source == nil { + return nil, 0, store.ErrSessionNotFound + } + + s.mu.RLock() + sourceSnapshot := *source + sourceMsgs := make([]providers.Message, len(source.Messages)) + copy(sourceMsgs, source.Messages) + sourceSnapshot.Messages = sourceMsgs + sourceMeta := make(map[string]string, len(source.Metadata)) + maps.Copy(sourceMeta, source.Metadata) + sourceSnapshot.Metadata = sourceMeta + s.mu.RUnlock() + + if opts.UpToIndex > len(sourceSnapshot.Messages) { + return nil, 0, store.ErrInvalidSessionBranch + } + + now := time.Now().UTC() + copied := make([]providers.Message, opts.UpToIndex) + copy(copied, sourceSnapshot.Messages[:opts.UpToIndex]) + meta := make(map[string]string, len(sourceSnapshot.Metadata)+len(opts.Metadata)+3) + maps.Copy(meta, sourceSnapshot.Metadata) + maps.Copy(meta, opts.Metadata) + meta["branched_from"] = sourceKey + meta["branched_from_index"] = strconv.Itoa(opts.UpToIndex) + meta["branched_at"] = now.Format(time.RFC3339Nano) + + branch := &store.SessionData{ + Key: opts.NewKey, + Messages: copied, + Summary: sourceSnapshot.Summary, + Created: now, + Updated: now, + AgentUUID: sourceSnapshot.AgentUUID, + UserID: sourceSnapshot.UserID, + Model: sourceSnapshot.Model, + Provider: sourceSnapshot.Provider, + Channel: "branch", + Label: opts.Label, + Metadata: meta, + ContextWindow: sourceSnapshot.ContextWindow, + } + if sourceSnapshot.TeamID != nil { + teamID := *sourceSnapshot.TeamID + branch.TeamID = &teamID + } + + msgsJSON, _ := json.Marshal(branch.Messages) + metaJSON := []byte("{}") + if len(branch.Metadata) > 0 { + metaJSON, _ = json.Marshal(branch.Metadata) + } + + res, err := s.db.ExecContext(ctx, + `INSERT INTO sessions (id, session_key, messages, summary, model, provider, channel, + input_tokens, output_tokens, compaction_count, + memory_flush_compaction_count, memory_flush_at, + label, spawned_by, spawn_depth, agent_id, user_id, metadata, updated_at, team_id, tenant_id, created_at) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) + ON CONFLICT(session_key, tenant_id) DO NOTHING`, + uuid.Must(uuid.NewV7()), branch.Key, msgsJSON, + nilStr(branch.Summary), nilStr(branch.Model), nilStr(branch.Provider), nilStr(branch.Channel), + branch.InputTokens, branch.OutputTokens, branch.CompactionCount, + branch.MemoryFlushCompactionCount, branch.MemoryFlushAt, + nilStr(branch.Label), nilStr(branch.SpawnedBy), branch.SpawnDepth, + nilSessionUUID(branch.AgentUUID), nilStr(branch.UserID), metaJSON, branch.Updated, + branch.TeamID, tenantIDForInsert(ctx), branch.Created, + ) + if err != nil { + return nil, 0, err + } + if n, _ := res.RowsAffected(); n == 0 { + return nil, 0, store.ErrSessionAlreadyExists + } + + s.mu.Lock() + s.cache[sessionCacheKey(ctx, opts.NewKey)] = branch + s.mu.Unlock() + return branch, len(copied), nil +} diff --git a/internal/store/sqlitestore/sessions_branch_test.go b/internal/store/sqlitestore/sessions_branch_test.go new file mode 100644 index 00000000..7709bae4 --- /dev/null +++ b/internal/store/sqlitestore/sessions_branch_test.go @@ -0,0 +1,114 @@ +//go:build sqlite || sqliteonly + +package sqlitestore + +import ( + "context" + "errors" + "testing" + + "github.com/nextlevelbuilder/goclaw/internal/providers" + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +func TestBranchSessionPersistsCopyAndMetadata(t *testing.T) { + db := openTestDB(t) + if err := EnsureSchema(db); err != nil { + t.Fatalf("EnsureSchema: %v", err) + } + + sessionStore := NewSQLiteSessionStore(db) + ctx := store.WithTenantID(context.Background(), store.MasterTenantID) + sourceKey := "agent:test-agent:direct:source-branch-test" + branchKey := "agent:test-agent:branch:copy-branch-test" + + source := sessionStore.GetOrCreate(ctx, sourceKey) + source.Summary = "source summary" + source.Metadata = map[string]string{"source": "yes"} + sessionStore.SetHistory(ctx, sourceKey, []providers.Message{ + {Role: "user", Content: "first"}, + {Role: "assistant", Content: "second"}, + }) + if err := sessionStore.Save(ctx, sourceKey); err != nil { + t.Fatalf("Save source: %v", err) + } + + branch, copied, err := sessionStore.BranchSession(ctx, sourceKey, store.SessionBranchOpts{ + NewKey: branchKey, + UpToIndex: 1, + Label: "branch label", + Metadata: map[string]string{"requested_by": "test"}, + }) + if err != nil { + t.Fatalf("BranchSession: %v", err) + } + if copied != 1 { + t.Fatalf("copied = %d, want 1", copied) + } + if branch.Key != branchKey || branch.Channel != "branch" || branch.Label != "branch label" { + t.Fatalf("branch metadata = key:%q channel:%q label:%q", branch.Key, branch.Channel, branch.Label) + } + if len(branch.Messages) != 1 || branch.Messages[0].Content != "first" { + t.Fatalf("branch messages = %+v", branch.Messages) + } + if branch.Metadata["branched_from"] != sourceKey || branch.Metadata["branched_from_index"] != "1" { + t.Fatalf("branch provenance metadata = %+v", branch.Metadata) + } + if branch.Metadata["source"] != "yes" || branch.Metadata["requested_by"] != "test" { + t.Fatalf("branch merged metadata = %+v", branch.Metadata) + } + + reloaded := NewSQLiteSessionStore(db).Get(ctx, branchKey) + if reloaded == nil { + t.Fatal("reloaded branch = nil") + } + if len(reloaded.Messages) != 1 || reloaded.Messages[0].Content != "first" { + t.Fatalf("reloaded messages = %+v", reloaded.Messages) + } + if reloaded.Metadata["branched_from"] != sourceKey { + t.Fatalf("reloaded metadata = %+v", reloaded.Metadata) + } +} + +func TestBranchSessionRejectsExistingTargetWithoutOverwrite(t *testing.T) { + db := openTestDB(t) + if err := EnsureSchema(db); err != nil { + t.Fatalf("EnsureSchema: %v", err) + } + + ctx := store.WithTenantID(context.Background(), store.MasterTenantID) + sourceKey := "agent:test-agent:direct:source-conflict-test" + branchKey := "agent:test-agent:branch:conflict-test" + + firstStore := NewSQLiteSessionStore(db) + firstStore.GetOrCreate(ctx, sourceKey) + firstStore.SetHistory(ctx, sourceKey, []providers.Message{{Role: "user", Content: "first"}}) + if err := firstStore.Save(ctx, sourceKey); err != nil { + t.Fatalf("Save source: %v", err) + } + if _, _, err := firstStore.BranchSession(ctx, sourceKey, store.SessionBranchOpts{ + NewKey: branchKey, + UpToIndex: 1, + Label: "first label", + }); err != nil { + t.Fatalf("BranchSession first: %v", err) + } + + secondStore := NewSQLiteSessionStore(db) + _, _, err := secondStore.BranchSession(ctx, sourceKey, store.SessionBranchOpts{ + NewKey: branchKey, + UpToIndex: 1, + Label: "second label", + }) + if !errors.Is(err, store.ErrSessionAlreadyExists) { + t.Fatalf("second BranchSession err = %v, want ErrSessionAlreadyExists", err) + } + + reloaded := NewSQLiteSessionStore(db).Get(ctx, branchKey) + if reloaded == nil { + t.Fatal("reloaded branch = nil") + } + if reloaded.Label != "first label" { + t.Fatalf("label = %q, want first label", reloaded.Label) + } +}