feat(http): unblock CLI P6 backend APIs (#44)

* feat(api): unblock trace follow and provider reconnect

* test(cron): stabilize run log execution test

* feat(http): add CLI P6 backend endpoints

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).

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.
 
 ---
 

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**

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
+}

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)
+	}
+}

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)

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,

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)
+	}
+}

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) {}

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 {

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)
+	}
+}

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)
+}

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"],

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 {

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)
+	}
+}

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"`
+}

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)
 }

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
+	}
+}

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
+}

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)

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
+	}
+}

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
+}

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)
+	}
+}