diff --git a/cmd/gateway.go b/cmd/gateway.go index 5e1adb59..f932f8a1 100644 --- a/cmd/gateway.go +++ b/cmd/gateway.go @@ -19,6 +19,7 @@ import ( "github.com/nextlevelbuilder/goclaw/internal/bootstrap" "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/cache" + "github.com/nextlevelbuilder/goclaw/internal/channelmemory" "github.com/nextlevelbuilder/goclaw/internal/channels" "github.com/nextlevelbuilder/goclaw/internal/channels/bitrix24" "github.com/nextlevelbuilder/goclaw/internal/channels/discord" @@ -260,6 +261,12 @@ func runGateway() { } } + if memorySvc := makeChannelMemoryService(pgStores, domainBus, providerRegistry, usageCapSvc); memorySvc != nil { + cleanupChannelMemory := (&channelmemory.Worker{Service: memorySvc}).Start(context.Background()) + defer cleanupChannelMemory() + slog.Info("channel memory extraction worker registered") + } + // V3: Wire vault enrichment worker (async summary + embedding + auto-linking). // Provider is resolved per-tenant at runtime — no static provider needed. var enrichProgress *vault.EnrichProgress @@ -383,7 +390,7 @@ func runGateway() { httpapi.InitGatewayNoAuthFallbackAllowed(config.GatewayNoAuthFallbackAllowed(cfg.Gateway)) exportTokenStore := httpapi.InitExportTokenStore() defer exportTokenStore.Stop() - agentsH, skillsH, tracesH, mcpH, channelInstancesH, providersH, builtinToolsH, pendingMessagesH, teamEventsH, secureCLIH, secureCLIGrantH, mcpUserCredsH := wireHTTP(pgStores, cfg.Agents.Defaults.Workspace, dataDir, bundledSkillsDir, msgBus, toolsReg, providerRegistry, modelReg, permPE.IsOwner, gatewayAddr, mcpToolLister, usageCapSvc, cfg, cfg.Skills) + agentsH, skillsH, tracesH, mcpH, channelInstancesH, providersH, builtinToolsH, pendingMessagesH, teamEventsH, secureCLIH, secureCLIGrantH, mcpUserCredsH := wireHTTP(pgStores, cfg.Agents.Defaults.Workspace, dataDir, bundledSkillsDir, msgBus, domainBus, toolsReg, providerRegistry, modelReg, permPE.IsOwner, gatewayAddr, mcpToolLister, usageCapSvc, cfg, cfg.Skills) // Wire dependencies for system prompt preview parity. if agentsH != nil { diff --git a/cmd/gateway_http_handlers.go b/cmd/gateway_http_handlers.go index 27962bc5..e47def21 100644 --- a/cmd/gateway_http_handlers.go +++ b/cmd/gateway_http_handlers.go @@ -2,7 +2,9 @@ package cmd import ( "github.com/nextlevelbuilder/goclaw/internal/bus" + "github.com/nextlevelbuilder/goclaw/internal/channelmemory" "github.com/nextlevelbuilder/goclaw/internal/config" + "github.com/nextlevelbuilder/goclaw/internal/eventbus" httpapi "github.com/nextlevelbuilder/goclaw/internal/http" "github.com/nextlevelbuilder/goclaw/internal/providers" "github.com/nextlevelbuilder/goclaw/internal/store" @@ -11,7 +13,7 @@ import ( ) // wireHTTP creates HTTP handlers (agents + skills + traces + MCP + channel instances + providers + builtin tools + pending messages). -func wireHTTP(stores *store.Stores, defaultWorkspace, dataDir, bundledSkillsDir string, msgBus *bus.MessageBus, toolsReg *tools.Registry, providerReg *providers.Registry, modelReg providers.ModelRegistry, isOwner func(string) bool, gatewayAddr string, mcpToolLister httpapi.MCPToolLister, usageCapSvc *usagecaps.Service, appCfg *config.Config, skillUploadConfig config.SkillsConfig) (*httpapi.AgentsHandler, *httpapi.SkillsHandler, *httpapi.TracesHandler, *httpapi.MCPHandler, *httpapi.ChannelInstancesHandler, *httpapi.ProvidersHandler, *httpapi.BuiltinToolsHandler, *httpapi.PendingMessagesHandler, *httpapi.TeamEventsHandler, *httpapi.SecureCLIHandler, *httpapi.SecureCLIGrantHandler, *httpapi.MCPUserCredentialsHandler) { +func wireHTTP(stores *store.Stores, defaultWorkspace, dataDir, bundledSkillsDir string, msgBus *bus.MessageBus, domainBus eventbus.DomainEventBus, toolsReg *tools.Registry, providerReg *providers.Registry, modelReg providers.ModelRegistry, isOwner func(string) bool, gatewayAddr string, mcpToolLister httpapi.MCPToolLister, usageCapSvc *usagecaps.Service, appCfg *config.Config, skillUploadConfig config.SkillsConfig) (*httpapi.AgentsHandler, *httpapi.SkillsHandler, *httpapi.TracesHandler, *httpapi.MCPHandler, *httpapi.ChannelInstancesHandler, *httpapi.ProvidersHandler, *httpapi.BuiltinToolsHandler, *httpapi.PendingMessagesHandler, *httpapi.TeamEventsHandler, *httpapi.SecureCLIHandler, *httpapi.SecureCLIGrantHandler, *httpapi.MCPUserCredentialsHandler) { var agentsH *httpapi.AgentsHandler var skillsH *httpapi.SkillsHandler var tracesH *httpapi.TracesHandler @@ -63,6 +65,9 @@ func wireHTTP(stores *store.Stores, defaultWorkspace, dataDir, bundledSkillsDir if stores != nil && stores.ChannelInstances != nil { channelInstancesH = httpapi.NewChannelInstancesHandler(stores.ChannelInstances, stores.Agents, stores.ConfigPermissions, stores.Contacts, stores.Tenants, msgBus) channelInstancesH.SetCapabilityStores(stores.MCP, stores.SecureCLI) + if memorySvc := makeChannelMemoryService(stores, domainBus, providerReg, usageCapSvc); memorySvc != nil { + channelInstancesH.SetMemoryExtractionService(memorySvc) + } } if stores != nil && stores.Providers != nil { @@ -115,3 +120,20 @@ func wireHTTP(stores *store.Stores, defaultWorkspace, dataDir, bundledSkillsDir return agentsH, skillsH, tracesH, mcpH, channelInstancesH, providersH, builtinToolsH, pendingMessagesH, teamEventsH, secureCLIH, secureCLIGrantH, mcpUserCredsH } + +func makeChannelMemoryService(stores *store.Stores, domainBus eventbus.DomainEventBus, providerReg *providers.Registry, usageCapSvc *usagecaps.Service) *channelmemory.Service { + if stores == nil || stores.ChannelInstances == nil || stores.PendingMessages == nil || stores.ChannelMemory == nil || stores.Episodic == nil { + return nil + } + return &channelmemory.Service{ + Channels: stores.ChannelInstances, + Pending: stores.PendingMessages, + Extractions: stores.ChannelMemory, + Episodic: stores.Episodic, + EventBus: domainBus, + SystemConfigs: stores.SystemConfigs, + Registry: providerReg, + UsageCaps: usageCapSvc, + Redactor: channelmemory.NewRedactor(), + } +} diff --git a/docs/05-channels-messaging.md b/docs/05-channels-messaging.md index 0b82ddfc..65e8fa46 100644 --- a/docs/05-channels-messaging.md +++ b/docs/05-channels-messaging.md @@ -644,7 +644,32 @@ Channel instances are loaded from the database with their assigned agent ID. The --- -## 13. Local Key Propagation +## 13. Passive Memory Extraction + +Passive channel memory is an opt-in per-channel feature. When enabled in +`channel_instances.config.passive_memory`, the gateway periodically reads the +existing tenant-scoped `channel_pending_messages` group buffers, redacts sensitive +content, asks the background LLM provider for durable fact candidates, and stores +only candidates in the review queue. + +Default behavior is privacy-first: + +- Disabled by default. +- Group-only in v1. +- Review mode enabled by default. +- Runs by manual trigger, message cap, or interval. +- New extraction tables store metadata, summaries, topics/entities, confidence, + status, and redaction counts, but not raw message bodies. + +Approved items write an `episodic_summaries` row with `source_type='channel'` +and a deterministic `source_id`; existing consolidation workers then handle KG +promotion. Reject/delete prevents later writes. Delete also removes the linked +episodic row when one exists; already-promoted KG nodes are not synchronously +deleted in v1. + +--- + +## 14. Local Key Propagation Thread/topic context is preserved through the entire message pipeline using a `local_key` in message metadata. This ensures subagent, delegation, and team message results land in the correct thread — not the root chat. @@ -660,7 +685,7 @@ All channel state — placeholders, streams, reactions, typing controllers, thre --- -## 14. Per-User Isolation +## 15. Per-User Isolation Channels provide per-user isolation through compound sender IDs and context propagation: @@ -671,7 +696,7 @@ Channels provide per-user isolation through compound sender IDs and context prop --- -## 15. Pairing System +## 16. Pairing System The pairing system provides a DM authentication flow for channels using the `pairing` DM policy. diff --git a/docs/06-store-data-model.md b/docs/06-store-data-model.md index bfbbbc8e..e71bc3a1 100644 --- a/docs/06-store-data-model.md +++ b/docs/06-store-data-model.md @@ -728,6 +728,13 @@ L0 (Working Memory) L1 (Episodic Memory) L2 (Semantic Memory) | `vault_versions` | Document version history (prepared for v3.1) | `doc_id`, `version`, `content`, `changed_by`, `created_at` | | `kg_entities` | Extended with temporal columns | `valid_from` (TIMESTAMPTZ), `valid_until` (TIMESTAMPTZ) for temporal facts | | `kg_relations` | Extended with temporal columns | `valid_from` (TIMESTAMPTZ), `valid_until` (TIMESTAMPTZ) for temporal edges | +| `channel_memory_extraction_runs` | Passive channel extraction run log | `tenant_id`, `channel_instance_id`, `history_key`, `trigger`, `status`, source range, counts, redaction metadata | +| `channel_memory_extraction_items` | Review queue for passive channel memory candidates | `tenant_id`, `run_id`, `channel_instance_id`, `item_hash`, `item_type`, `summary`, `topics`, `entities`, `status`, approval/write timestamps | + +`ChannelMemoryExtractionStore` is implemented for PostgreSQL and SQLite. It is +tenant-scoped, stores no raw message bodies, and uses deterministic hashes to +deduplicate the same channel/history/type/summary candidate across repeated +runs. ### 12 Promoted Agent Columns diff --git a/docs/07-bootstrap-skills-memory.md b/docs/07-bootstrap-skills-memory.md index 62c721a0..242052ac 100644 --- a/docs/07-bootstrap-skills-memory.md +++ b/docs/07-bootstrap-skills-memory.md @@ -596,6 +596,13 @@ flowchart LR 7. Stores in `episodic_summaries` 8. Publishes `episodic.created` for downstream workers +**Passive channel memory** (`internal/channelmemory`): +1. Reads existing channel pending-message groups only when a channel admin enables `passive_memory.enabled` +2. Redacts secrets, tokens, connection strings, payment-like numbers, emails, phones, and configured excluded users/patterns +3. Writes extracted candidates to `channel_memory_extraction_items` for review by default +4. On approval, creates an `episodic_summaries` row with `source_type='channel'` +5. Publishes `episodic.created` so SemanticWorker/DedupWorker use the same KG path as session memory + **SemanticWorker** (`internal/consolidation/semantic_worker.go`): 1. Listens to `episodic.created` events 2. Parses summary for entity mentions + relationships diff --git a/docs/09-security.md b/docs/09-security.md index e4384f1c..eb3c22be 100644 --- a/docs/09-security.md +++ b/docs/09-security.md @@ -129,6 +129,16 @@ All four filesystem tools (`read_file`, `write_file`, `list_files`, `edit`) impl | Dynamic value scrubbing (SSRF) | Server IPs and other runtime-discovered values registered via `AddDynamicScrubValues()` and replaced with `[SERVER_IP]` | | Web content wrapping | Fetched content wrapped in `<<>>` tags with security warning | +### Passive Channel Memory Redaction + +Passive channel memory extraction runs an additional pre-LLM redaction pass over +pending group messages before any durable candidate is created. The pass removes +common secret/key assignments, bearer/provider tokens, connection strings, +payment-like digit sequences, phone numbers, email addresses, configured +excluded sender IDs, and configured regex exclusion patterns. The new extraction +tables store redaction counts/types and extracted summaries only, not raw message +bodies. + ### Layer 5: Isolation **Per-user workspace isolation** -- Two levels prevent cross-user file access: diff --git a/docs/18-http-api.md b/docs/18-http-api.md index 44bedddf..ae7fc6cd 100644 --- a/docs/18-http-api.md +++ b/docs/18-http-api.md @@ -1111,6 +1111,28 @@ Response includes `allowed`, `reason`, `instance_id`, `agent_id`, `group_id`, `u Credentials are masked in HTTP responses. +### Passive Memory Extraction + +Tenant admins can configure and operate passive memory extraction per channel +instance. Read endpoints are viewer-accessible; settings and item transitions +require tenant admin. + +| Method | Path | Description | +|--------|------|-------------| +| `GET` | `/v1/channels/instances/{id}/memory-extraction` | Get config, latest run, pending count, and recent items | +| `PUT` | `/v1/channels/instances/{id}/memory-extraction/settings` | Replace normalized `passive_memory` config in the channel instance config | +| `POST` | `/v1/channels/instances/{id}/memory-extraction/run` | Trigger a manual extraction run | +| `GET` | `/v1/channels/instances/{id}/memory-extraction/items` | List review queue items; optional `status` filter | +| `POST` | `/v1/channels/instances/{id}/memory-extraction/items/{itemID}/approve` | Write candidate to episodic memory and publish KG event | +| `POST` | `/v1/channels/instances/{id}/memory-extraction/items/{itemID}/reject` | Reject candidate | +| `DELETE` | `/v1/channels/instances/{id}/memory-extraction/items/{itemID}` | Delete candidate and linked episodic summary when present | + +Config fields: `enabled`, `review_mode`, `interval_minutes`, `message_cap`, +`retention_hours`, `allowed_types`, `exclude_users`, `exclude_patterns`, +`min_messages`, `group_only`. Defaults are disabled, review mode on, group-only, +360 minute interval, 100 message cap, 168 hour retention, and durable types: +people, projects, decisions, todos, preferences, events. + --- ## 18. Pending Messages diff --git a/docs/project-changelog.md b/docs/project-changelog.md index a6d18592..d8a84842 100644 --- a/docs/project-changelog.md +++ b/docs/project-changelog.md @@ -6,6 +6,22 @@ Significant changes, features, and fixes in reverse chronological order. ## 2026-05-29 +### Passive channel memory extraction (issue #64) + +**New** + +- Added opt-in per-channel passive memory extraction from pending group buffers, with scheduled/manual runs, redaction before LLM extraction, review queue items, and deterministic channel memory source IDs. +- Added PostgreSQL migration `000076` and SQLite schema version `45` for `channel_memory_extraction_runs` and `channel_memory_extraction_items`. +- Added tenant-admin HTTP APIs and channel detail UI controls for settings, manual run, last run status, and approve/reject/delete review actions. + +**Security** + +- Feature is disabled by default and group-only in v1. +- Review mode is on by default; approved items write to `episodic_summaries` with `source_type='channel'` and then enter the existing KG consolidation event path. +- Delete removes the review item and associated episodic summary when present. Existing KG nodes created before deletion may require later dedup/cleanup; raw channel messages are not copied into the new extraction tables. + +--- + ### Channel context admin surface (issue #66) **New** diff --git a/internal/channelmemory/config.go b/internal/channelmemory/config.go new file mode 100644 index 00000000..e3604941 --- /dev/null +++ b/internal/channelmemory/config.go @@ -0,0 +1,126 @@ +package channelmemory + +import ( + "encoding/json" + "regexp" + "slices" + "time" +) + +var DefaultAllowedTypes = []string{"people", "projects", "decisions", "todos", "preferences", "events"} + +type Config struct { + Enabled bool `json:"enabled"` + ReviewMode bool `json:"review_mode"` + IntervalMinutes int `json:"interval_minutes"` + MessageCap int `json:"message_cap"` + RetentionHours int `json:"retention_hours"` + AllowedTypes []string `json:"allowed_types"` + ExcludeUsers []string `json:"exclude_users"` + ExcludePatterns []string `json:"exclude_patterns"` + MinMessages int `json:"min_messages"` + GroupOnly bool `json:"group_only"` +} + +func DefaultConfig() Config { + return Config{ + Enabled: false, + ReviewMode: true, + IntervalMinutes: 360, + MessageCap: 100, + RetentionHours: 168, + AllowedTypes: slices.Clone(DefaultAllowedTypes), + MinMessages: 5, + GroupOnly: true, + } +} + +func ParseConfig(raw json.RawMessage) Config { + cfg := DefaultConfig() + if len(raw) == 0 { + return cfg + } + var root struct { + PassiveMemory *Config `json:"passive_memory"` + } + if err := json.Unmarshal(raw, &root); err != nil || root.PassiveMemory == nil { + return cfg + } + in := root.PassiveMemory + cfg.Enabled = in.Enabled + cfg.ReviewMode = in.ReviewMode + cfg.IntervalMinutes = clampInt(in.IntervalMinutes, 15, 10080, cfg.IntervalMinutes) + cfg.MessageCap = clampInt(in.MessageCap, 10, 1000, cfg.MessageCap) + cfg.RetentionHours = clampInt(in.RetentionHours, 1, 720, cfg.RetentionHours) + cfg.AllowedTypes = normalizeAllowedTypes(in.AllowedTypes) + cfg.ExcludeUsers = boundedStrings(in.ExcludeUsers, 50, 255) + cfg.ExcludePatterns = boundedPatterns(in.ExcludePatterns, 20, 255) + cfg.MinMessages = clampInt(in.MinMessages, 2, 100, cfg.MinMessages) + cfg.GroupOnly = true + return cfg +} + +func MergeIntoInstanceConfig(raw json.RawMessage, cfg Config) json.RawMessage { + var root map[string]any + if len(raw) == 0 || json.Unmarshal(raw, &root) != nil || root == nil { + root = make(map[string]any) + } + root["passive_memory"] = cfg + out, _ := json.Marshal(root) + return out +} + +func (c Config) Interval() time.Duration { + return time.Duration(c.IntervalMinutes) * time.Minute +} + +func normalizeAllowedTypes(in []string) []string { + allowed := make([]string, 0, len(DefaultAllowedTypes)) + for _, v := range in { + if slices.Contains(DefaultAllowedTypes, v) && !slices.Contains(allowed, v) { + allowed = append(allowed, v) + } + } + if len(allowed) == 0 { + return slices.Clone(DefaultAllowedTypes) + } + return allowed +} + +func boundedStrings(in []string, maxCount, maxLen int) []string { + out := make([]string, 0, min(len(in), maxCount)) + for _, v := range in { + if v == "" || len(v) > maxLen { + continue + } + out = append(out, v) + if len(out) >= maxCount { + break + } + } + return out +} + +func boundedPatterns(in []string, maxCount, maxLen int) []string { + candidates := boundedStrings(in, maxCount, maxLen) + out := candidates[:0] + for _, v := range candidates { + if _, err := regexp.Compile(v); err == nil { + out = append(out, v) + } + } + return out +} + +func clampInt(v, minV, maxV, fallback int) int { + if v == 0 { + return fallback + } + if v < minV { + return minV + } + if v > maxV { + return maxV + } + return v +} diff --git a/internal/channelmemory/config_test.go b/internal/channelmemory/config_test.go new file mode 100644 index 00000000..b92ef44c --- /dev/null +++ b/internal/channelmemory/config_test.go @@ -0,0 +1,81 @@ +package channelmemory + +import ( + "encoding/json" + "slices" + "testing" +) + +func TestParseConfigDefaultsDisabled(t *testing.T) { + cfg := ParseConfig(nil) + if cfg.Enabled { + t.Fatal("default passive memory must be disabled") + } + if !cfg.ReviewMode { + t.Fatal("default passive memory must require review") + } + if !slices.Equal(cfg.AllowedTypes, DefaultAllowedTypes) { + t.Fatalf("allowed types = %v, want %v", cfg.AllowedTypes, DefaultAllowedTypes) + } +} + +func TestParseConfigNormalizesBoundsAndTypes(t *testing.T) { + raw := json.RawMessage(`{ + "passive_memory": { + "enabled": true, + "review_mode": true, + "interval_minutes": 2, + "message_cap": 5000, + "retention_hours": 0, + "allowed_types": ["people", "people", "unknown", "todos"], + "exclude_users": ["u1", ""], + "exclude_patterns": ["secret", "["], + "min_messages": 1, + "group_only": false + } + }`) + cfg := ParseConfig(raw) + if !cfg.Enabled || !cfg.ReviewMode { + t.Fatalf("unexpected enabled/review flags: %+v", cfg) + } + if cfg.IntervalMinutes != 15 { + t.Fatalf("interval = %d, want lower bound 15", cfg.IntervalMinutes) + } + if cfg.MessageCap != 1000 { + t.Fatalf("message cap = %d, want upper bound 1000", cfg.MessageCap) + } + if cfg.RetentionHours != 168 { + t.Fatalf("retention = %d, want fallback 168", cfg.RetentionHours) + } + if !slices.Equal(cfg.AllowedTypes, []string{"people", "todos"}) { + t.Fatalf("allowed types = %v", cfg.AllowedTypes) + } + if !slices.Equal(cfg.ExcludeUsers, []string{"u1"}) { + t.Fatalf("exclude users = %v", cfg.ExcludeUsers) + } + if !slices.Equal(cfg.ExcludePatterns, []string{"secret"}) { + t.Fatalf("exclude patterns = %v", cfg.ExcludePatterns) + } + if cfg.MinMessages != 2 { + t.Fatalf("min messages = %d, want lower bound 2", cfg.MinMessages) + } + if !cfg.GroupOnly { + t.Fatal("group_only must normalize to true") + } +} + +func TestMergeIntoInstanceConfigPreservesSiblingFields(t *testing.T) { + cfg := DefaultConfig() + cfg.Enabled = true + raw := MergeIntoInstanceConfig(json.RawMessage(`{"foo":"bar"}`), cfg) + var root map[string]any + if err := json.Unmarshal(raw, &root); err != nil { + t.Fatal(err) + } + if root["foo"] != "bar" { + t.Fatalf("sibling field missing: %s", raw) + } + if _, ok := root["passive_memory"]; !ok { + t.Fatalf("passive_memory missing: %s", raw) + } +} diff --git a/internal/channelmemory/extractor.go b/internal/channelmemory/extractor.go new file mode 100644 index 00000000..acd3017a --- /dev/null +++ b/internal/channelmemory/extractor.go @@ -0,0 +1,107 @@ +package channelmemory + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "time" + + "github.com/nextlevelbuilder/goclaw/internal/providers" + "github.com/nextlevelbuilder/goclaw/internal/store" + usagecaps "github.com/nextlevelbuilder/goclaw/internal/usage/caps" +) + +type ExtractedItem struct { + Type string `json:"type"` + Summary string `json:"summary"` + Topics []string `json:"topics"` + Entities []string `json:"entities"` + Confidence float64 `json:"confidence"` +} + +func Extract(ctx context.Context, provider providers.Provider, model string, caps *usagecaps.Service, messages []store.PendingMessage, allowed []string) ([]ExtractedItem, error) { + if provider == nil { + return nil, fmt.Errorf("background provider unavailable") + } + var sb strings.Builder + for _, msg := range messages { + sb.WriteString(msg.CreatedAt.Format(time.RFC3339)) + sb.WriteString(" ") + if msg.Sender != "" { + sb.WriteString(msg.Sender) + } else { + sb.WriteString(msg.SenderID) + } + sb.WriteString(": ") + body := msg.Body + if len([]rune(body)) > 800 { + body = string([]rune(body)[:800]) + "..." + } + sb.WriteString(body) + sb.WriteByte('\n') + if sb.Len() > 12000 { + sb.WriteString("...(truncated)\n") + break + } + } + req := providers.ChatRequest{ + Model: model, + Messages: []providers.Message{ + {Role: "system", Content: extractionPrompt(allowed)}, + {Role: "user", Content: sb.String()}, + }, + Options: map[string]any{"max_tokens": 1200, "temperature": 0.1}, + } + var resp *providers.ChatResponse + var err error + if caps != nil { + resp, err = caps.Chat(ctx, provider, req, usagecaps.ChatOptions{ + ModelID: model, + Purpose: "channel-memory-extraction", + MaxOutputTokens: 1200, + }) + } else { + resp, err = provider.Chat(ctx, req) + } + if err != nil { + return nil, err + } + return parseExtractionResponse(resp.Content) +} + +func extractionPrompt(allowed []string) string { + return `Extract only durable, reusable work context from channel messages. +Allowed item types: ` + strings.Join(allowed, ", ") + `. +Never include secrets, credentials, tokens, payment data, private addresses, phone numbers, health/legal/financial sensitive details, casual chatter, jokes, or low-confidence guesses. +Return strict JSON array only. Each item: +{"type":"people|projects|decisions|todos|preferences|events","summary":"one concise redacted fact","topics":["..."],"entities":["..."],"confidence":0.0-1.0} +If nothing durable remains, return [].` +} + +func parseExtractionResponse(content string) ([]ExtractedItem, error) { + raw := strings.TrimSpace(content) + raw = strings.TrimPrefix(raw, "```json") + raw = strings.TrimPrefix(raw, "```") + raw = strings.TrimSuffix(raw, "```") + raw = strings.TrimSpace(raw) + var items []ExtractedItem + if err := json.Unmarshal([]byte(raw), &items); err != nil { + return nil, fmt.Errorf("parse extraction JSON: %w", err) + } + out := items[:0] + for _, item := range items { + item.Summary = strings.TrimSpace(item.Summary) + if item.Summary == "" || item.Type == "" { + continue + } + if item.Confidence < 0 { + item.Confidence = 0 + } + if item.Confidence > 1 { + item.Confidence = 1 + } + out = append(out, item) + } + return out, nil +} diff --git a/internal/channelmemory/redactor.go b/internal/channelmemory/redactor.go new file mode 100644 index 00000000..65db2d8e --- /dev/null +++ b/internal/channelmemory/redactor.go @@ -0,0 +1,91 @@ +package channelmemory + +import ( + "regexp" + "slices" + "strings" + + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +type RedactionResult struct { + Messages []store.PendingMessage + Count int + Types []string +} + +type Redactor struct { + patterns []namedPattern +} + +type namedPattern struct { + name string + re *regexp.Regexp +} + +func NewRedactor() *Redactor { + return &Redactor{patterns: []namedPattern{ + {"secret", regexp.MustCompile(`(?i)\b(api[_-]?key|token|secret|password|passwd|pwd)\b\s*[:=]\s*['"]?[^'" \n\r]+`)}, + {"token", regexp.MustCompile(`(?i)\b(bearer|sk-[a-z0-9_-]{12,}|ghp_[a-z0-9_]+|xox[baprs]-[a-z0-9-]+)\b[^\s]*`)}, + {"connection_string", regexp.MustCompile(`(?i)\b(postgres|postgresql|mysql|redis|mongodb)://[^\s]+`)}, + {"payment", regexp.MustCompile(`\b(?:\d[ -]*?){13,19}\b`)}, + {"phone", regexp.MustCompile(`(?i)(\+?\d[\d .()-]{7,}\d)`)}, + {"email", regexp.MustCompile(`\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b`)}, + }} +} + +func (r *Redactor) Redact(messages []store.PendingMessage, cfg Config) RedactionResult { + excludePatterns := compileUserPatterns(cfg.ExcludePatterns) + out := make([]store.PendingMessage, 0, len(messages)) + types := make([]string, 0) + count := 0 + for _, msg := range messages { + if slices.Contains(cfg.ExcludeUsers, msg.SenderID) { + count++ + addType(&types, "excluded_user") + continue + } + body := msg.Body + excluded := false + for _, re := range excludePatterns { + if re.MatchString(body) { + excluded = true + count++ + addType(&types, "excluded_pattern") + break + } + } + if excluded { + continue + } + for _, p := range r.patterns { + next := p.re.ReplaceAllString(body, "[REDACTED:"+p.name+"]") + if next != body { + count++ + addType(&types, p.name) + body = next + } + } + msg.Body = strings.TrimSpace(body) + if msg.Body != "" { + out = append(out, msg) + } + } + return RedactionResult{Messages: out, Count: count, Types: types} +} + +func compileUserPatterns(patterns []string) []*regexp.Regexp { + out := make([]*regexp.Regexp, 0, len(patterns)) + for _, pattern := range patterns { + if re, err := regexp.Compile(pattern); err == nil { + out = append(out, re) + } + } + return out +} + +func addType(types *[]string, typ string) { + if !slices.Contains(*types, typ) { + *types = append(*types, typ) + } +} diff --git a/internal/channelmemory/redactor_test.go b/internal/channelmemory/redactor_test.go new file mode 100644 index 00000000..bc504d6c --- /dev/null +++ b/internal/channelmemory/redactor_test.go @@ -0,0 +1,49 @@ +package channelmemory + +import ( + "strings" + "testing" + + "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +func TestRedactorRedactsSecretsAndEmails(t *testing.T) { + msgs := []store.PendingMessage{{ + ID: uuid.New(), + SenderID: "u1", + Body: "email me at owner@example.com token: sk-1234567890abcdef", + }} + got := NewRedactor().Redact(msgs, DefaultConfig()) + if len(got.Messages) != 1 { + t.Fatalf("messages len = %d", len(got.Messages)) + } + body := got.Messages[0].Body + if strings.Contains(body, "owner@example.com") || strings.Contains(body, "sk-1234567890abcdef") { + t.Fatalf("sensitive content not redacted: %q", body) + } + if !strings.Contains(body, "[REDACTED:email]") || !strings.Contains(body, "[REDACTED:secret]") { + t.Fatalf("redaction markers missing: %q", body) + } +} + +func TestRedactorExcludesUsersAndPatterns(t *testing.T) { + cfg := DefaultConfig() + cfg.ExcludeUsers = []string{"ignored"} + cfg.ExcludePatterns = []string{"do not remember"} + msgs := []store.PendingMessage{ + {ID: uuid.New(), SenderID: "ignored", Body: "project secret"}, + {ID: uuid.New(), SenderID: "kept", Body: "do not remember this"}, + {ID: uuid.New(), SenderID: "kept", Body: "remember this decision"}, + } + got := NewRedactor().Redact(msgs, cfg) + if len(got.Messages) != 1 { + t.Fatalf("messages len = %d, want 1", len(got.Messages)) + } + if got.Messages[0].Body != "remember this decision" { + t.Fatalf("kept body = %q", got.Messages[0].Body) + } + if got.Count != 2 { + t.Fatalf("redaction count = %d, want 2", got.Count) + } +} diff --git a/internal/channelmemory/review.go b/internal/channelmemory/review.go new file mode 100644 index 00000000..cb8076e5 --- /dev/null +++ b/internal/channelmemory/review.go @@ -0,0 +1,102 @@ +package channelmemory + +import ( + "context" + "fmt" + "time" + + "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/eventbus" + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +func (s *Service) Approve(ctx context.Context, itemID uuid.UUID, approver string) (*store.ChannelMemoryExtractionItem, error) { + item, err := s.Extractions.GetItem(ctx, itemID) + if err != nil { + return nil, err + } + if item.Status == store.ChannelMemoryItemWritten { + return item, nil + } + if item.Status != store.ChannelMemoryItemPendingReview && item.Status != store.ChannelMemoryItemApproved { + return nil, fmt.Errorf("item is not approvable") + } + sourceID := item.SourceID + exists, err := s.Episodic.ExistsBySourceID(ctx, item.AgentID.String(), item.UserID, sourceID) + if err != nil { + return nil, err + } + if !exists { + ep := &store.EpisodicSummary{ + TenantID: item.TenantID, + AgentID: item.AgentID, + UserID: item.UserID, + SessionKey: "channel:" + item.ChannelInstanceID.String(), + Summary: item.Summary, + KeyTopics: decodeStrings(item.Topics), + SourceID: sourceID, + SourceType: "channel", + ExpiresAt: new(time.Now().UTC().Add(90 * 24 * time.Hour)), + } + if err := s.Episodic.Create(ctx, ep); err != nil { + return nil, err + } + item.EpisodicID = ep.ID.String() + if s.EventBus != nil { + s.EventBus.Publish(eventbus.DomainEvent{ + Type: eventbus.EventEpisodicCreated, + SourceID: ep.ID.String(), + TenantID: item.TenantID.String(), + AgentID: item.AgentID.String(), + UserID: item.UserID, + Payload: &eventbus.EpisodicCreatedPayload{ + EpisodicID: ep.ID.String(), + SessionKey: ep.SessionKey, + Summary: item.Summary, + KeyEntities: decodeStrings(item.Entities), + }, + }) + } + } + now := time.Now().UTC() + if err := s.Extractions.UpdateItem(ctx, item.ID, map[string]any{ + "status": store.ChannelMemoryItemWritten, + "approved_by": approver, + "approved_at": now, + "written_at": now, + "episodic_id": item.EpisodicID, + }); err != nil { + return nil, err + } + item.Status = store.ChannelMemoryItemWritten + return item, nil +} + +func (s *Service) Reject(ctx context.Context, itemID uuid.UUID, actor string) error { + return s.updateItemTerminal(ctx, itemID, store.ChannelMemoryItemRejected, map[string]any{"rejected_by": actor, "rejected_at": time.Now().UTC()}) +} + +func (s *Service) Delete(ctx context.Context, itemID uuid.UUID) error { + item, err := s.Extractions.GetItem(ctx, itemID) + if err != nil { + return err + } + if item.EpisodicID != "" && s.Episodic != nil { + if err := s.Episodic.Delete(ctx, item.EpisodicID); err != nil { + return err + } + } + return s.updateItemTerminal(ctx, itemID, store.ChannelMemoryItemDeleted, map[string]any{"deleted_at": time.Now().UTC()}) +} + +func (s *Service) updateItemTerminal(ctx context.Context, itemID uuid.UUID, status string, updates map[string]any) error { + item, err := s.Extractions.GetItem(ctx, itemID) + if err != nil { + return err + } + if item.Status == store.ChannelMemoryItemWritten && status != store.ChannelMemoryItemDeleted { + return fmt.Errorf("written item cannot transition to %s", status) + } + updates["status"] = status + return s.Extractions.UpdateItem(ctx, itemID, updates) +} diff --git a/internal/channelmemory/service.go b/internal/channelmemory/service.go new file mode 100644 index 00000000..79296afe --- /dev/null +++ b/internal/channelmemory/service.go @@ -0,0 +1,163 @@ +package channelmemory + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/eventbus" + "github.com/nextlevelbuilder/goclaw/internal/providerresolve" + "github.com/nextlevelbuilder/goclaw/internal/providers" + "github.com/nextlevelbuilder/goclaw/internal/store" + usagecaps "github.com/nextlevelbuilder/goclaw/internal/usage/caps" +) + +type Service struct { + Channels store.ChannelInstanceStore + Pending store.PendingMessageStore + Extractions store.ChannelMemoryExtractionStore + Episodic store.EpisodicStore + EventBus eventbus.DomainEventBus + SystemConfigs store.SystemConfigStore + Registry *providers.Registry + UsageCaps *usagecaps.Service + Redactor *Redactor +} + +type Status struct { + Config Config `json:"config"` + LastRun *store.ChannelMemoryExtractionRun `json:"last_run,omitempty"` + PendingCount int `json:"pending_count"` + RecentItems []store.ChannelMemoryExtractionItem `json:"recent_items"` +} + +func (s *Service) Status(ctx context.Context, inst *store.ChannelInstanceData) (*Status, error) { + cfg := ParseConfig(inst.Config) + runs, err := s.Extractions.ListRuns(ctx, store.ChannelMemoryRunListOptions{ChannelInstanceID: inst.ID, Limit: 1}) + if err != nil { + return nil, err + } + items, err := s.Extractions.ListItems(ctx, store.ChannelMemoryItemListOptions{ChannelInstanceID: inst.ID, Limit: 25}) + if err != nil { + return nil, err + } + pending := 0 + for _, item := range items { + if item.Status == store.ChannelMemoryItemPendingReview { + pending++ + } + } + var last *store.ChannelMemoryExtractionRun + if len(runs) > 0 { + last = &runs[0] + } + return &Status{Config: cfg, LastRun: last, PendingCount: pending, RecentItems: items}, nil +} + +func (s *Service) RunNow(ctx context.Context, inst *store.ChannelInstanceData, trigger string) (*store.ChannelMemoryExtractionRun, error) { + cfg := ParseConfig(inst.Config) + if !cfg.Enabled && trigger != "manual" { + return nil, fmt.Errorf("passive memory disabled") + } + groups, err := s.Pending.ListGroups(ctx) + if err != nil { + return nil, err + } + for _, group := range groups { + if group.ChannelName != inst.Name || !eligibleHistoryKey(group.HistoryKey, cfg) { + continue + } + if trigger != "manual" && !s.shouldRunScheduled(ctx, inst.ID, group, cfg) { + continue + } + return s.runGroup(ctx, inst, cfg, group, trigger) + } + return nil, fmt.Errorf("no eligible channel messages") +} + +func (s *Service) shouldRunScheduled(ctx context.Context, instID uuid.UUID, group store.PendingMessageGroup, cfg Config) bool { + if group.MessageCount >= cfg.MessageCap { + return true + } + runs, err := s.Extractions.ListRuns(ctx, store.ChannelMemoryRunListOptions{ + ChannelInstanceID: instID, + HistoryKey: group.HistoryKey, + Status: store.ChannelMemoryRunCompleted, + Limit: 1, + }) + if err != nil || len(runs) == 0 { + return true + } + return time.Since(runs[0].CreatedAt) >= cfg.Interval() +} + +func (s *Service) runGroup(ctx context.Context, inst *store.ChannelInstanceData, cfg Config, group store.PendingMessageGroup, trigger string) (*store.ChannelMemoryExtractionRun, error) { + messages, err := s.Pending.ListByKey(ctx, group.ChannelName, group.HistoryKey) + if err != nil { + return nil, err + } + if len(messages) < cfg.MinMessages { + return nil, fmt.Errorf("not enough useful messages") + } + redactor := s.Redactor + if redactor == nil { + redactor = NewRedactor() + } + redacted := redactor.Redact(messages, cfg) + if len(redacted.Messages) < cfg.MinMessages { + return nil, fmt.Errorf("not enough redacted messages") + } + start, end := redacted.Messages[0], redacted.Messages[len(redacted.Messages)-1] + redactionTypes, _ := json.Marshal(redacted.Types) + run := &store.ChannelMemoryExtractionRun{ + ChannelInstanceID: inst.ID, + ChannelName: inst.Name, + AgentID: inst.AgentID, + UserID: inst.CreatedBy, + HistoryKey: group.HistoryKey, + Trigger: trigger, + Status: store.ChannelMemoryRunRunning, + SourceStartID: messageSourceID(start), + SourceEndID: messageSourceID(end), + SourceStartAt: &start.CreatedAt, + SourceEndAt: &end.CreatedAt, + MessageCount: len(redacted.Messages), + RedactionCount: redacted.Count, + RedactionTypes: redactionTypes, + StartedAt: new(time.Now().UTC()), + } + if err := s.Extractions.CreateRun(ctx, run); err != nil { + return nil, err + } + provider, model := providerresolve.ResolveBackgroundProvider(ctx, run.TenantID, s.Registry, s.SystemConfigs) + items, err := Extract(ctx, provider, model, s.UsageCaps, redacted.Messages, cfg.AllowedTypes) + if err != nil { + _ = s.Extractions.UpdateRun(ctx, run.ID, map[string]any{ + "status": store.ChannelMemoryRunFailed, "error_message": err.Error(), "completed_at": time.Now().UTC(), + }) + return run, err + } + for _, extracted := range items { + if !contains(cfg.AllowedTypes, extracted.Type) { + continue + } + item := s.itemFromExtracted(run, extracted) + if err := s.Extractions.CreateItem(ctx, item); err != nil { + return run, err + } + if !cfg.ReviewMode { + if _, err := s.Approve(ctx, item.ID, "system"); err != nil { + return run, err + } + } + run.ItemCount++ + } + status := store.ChannelMemoryRunCompleted + _ = s.Extractions.UpdateRun(ctx, run.ID, map[string]any{ + "status": status, "item_count": run.ItemCount, "completed_at": time.Now().UTC(), + }) + run.Status = status + return run, nil +} diff --git a/internal/channelmemory/service_test.go b/internal/channelmemory/service_test.go new file mode 100644 index 00000000..86794aa5 --- /dev/null +++ b/internal/channelmemory/service_test.go @@ -0,0 +1,95 @@ +package channelmemory + +import ( + "context" + "database/sql" + "testing" + "time" + + "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +type fakeExtractionStore struct { + runs []store.ChannelMemoryExtractionRun +} + +func (f *fakeExtractionStore) CreateRun(context.Context, *store.ChannelMemoryExtractionRun) error { + return nil +} + +func (f *fakeExtractionStore) GetRun(context.Context, uuid.UUID) (*store.ChannelMemoryExtractionRun, error) { + return nil, sql.ErrNoRows +} + +func (f *fakeExtractionStore) ListRuns(context.Context, store.ChannelMemoryRunListOptions) ([]store.ChannelMemoryExtractionRun, error) { + return f.runs, nil +} + +func (f *fakeExtractionStore) UpdateRun(context.Context, uuid.UUID, map[string]any) error { + return nil +} + +func (f *fakeExtractionStore) CreateItem(context.Context, *store.ChannelMemoryExtractionItem) error { + return nil +} + +func (f *fakeExtractionStore) GetItem(context.Context, uuid.UUID) (*store.ChannelMemoryExtractionItem, error) { + return nil, sql.ErrNoRows +} + +func (f *fakeExtractionStore) ListItems(context.Context, store.ChannelMemoryItemListOptions) ([]store.ChannelMemoryExtractionItem, error) { + return nil, nil +} + +func (f *fakeExtractionStore) UpdateItem(context.Context, uuid.UUID, map[string]any) error { + return nil +} + +func TestShouldRunScheduledWhenMessageCapReached(t *testing.T) { + cfg := DefaultConfig() + cfg.MessageCap = 10 + cfg.IntervalMinutes = 360 + svc := &Service{Extractions: &fakeExtractionStore{runs: []store.ChannelMemoryExtractionRun{{ + CreatedAt: time.Now().UTC(), + }}}} + if !svc.shouldRunScheduled(context.Background(), uuid.New(), store.PendingMessageGroup{MessageCount: 10}, cfg) { + t.Fatal("expected scheduled run at message cap") + } +} + +func TestShouldRunScheduledWhenIntervalElapsed(t *testing.T) { + cfg := DefaultConfig() + cfg.MessageCap = 100 + cfg.IntervalMinutes = 60 + svc := &Service{Extractions: &fakeExtractionStore{runs: []store.ChannelMemoryExtractionRun{{ + CreatedAt: time.Now().UTC().Add(-2 * time.Hour), + }}}} + if !svc.shouldRunScheduled(context.Background(), uuid.New(), store.PendingMessageGroup{MessageCount: 20}, cfg) { + t.Fatal("expected scheduled run after interval") + } +} + +func TestShouldSkipScheduledBelowCapAndBeforeInterval(t *testing.T) { + cfg := DefaultConfig() + cfg.MessageCap = 100 + cfg.IntervalMinutes = 60 + svc := &Service{Extractions: &fakeExtractionStore{runs: []store.ChannelMemoryExtractionRun{{ + CreatedAt: time.Now().UTC(), + }}}} + if svc.shouldRunScheduled(context.Background(), uuid.New(), store.PendingMessageGroup{MessageCount: 20}, cfg) { + t.Fatal("expected scheduled run to wait") + } +} + +func TestItemHashIsStableAcrossRuns(t *testing.T) { + runA := &store.ChannelMemoryExtractionRun{ID: uuid.New(), ChannelInstanceID: uuid.New(), HistoryKey: "group"} + runB := *runA + runB.ID = uuid.New() + svc := &Service{} + itemA := svc.itemFromExtracted(runA, ExtractedItem{Type: "decision", Summary: "Ship beta"}) + itemB := svc.itemFromExtracted(&runB, ExtractedItem{Type: "decision", Summary: "Ship beta"}) + if itemA.ItemHash != itemB.ItemHash { + t.Fatalf("hash changed across runs: %s != %s", itemA.ItemHash, itemB.ItemHash) + } +} diff --git a/internal/channelmemory/util.go b/internal/channelmemory/util.go new file mode 100644 index 00000000..af692544 --- /dev/null +++ b/internal/channelmemory/util.go @@ -0,0 +1,60 @@ +package channelmemory + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "slices" + "strings" + "time" + + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +func (s *Service) itemFromExtracted(run *store.ChannelMemoryExtractionRun, extracted ExtractedItem) *store.ChannelMemoryExtractionItem { + topics, _ := json.Marshal(extracted.Topics) + entities, _ := json.Marshal(extracted.Entities) + hash := sha256.Sum256([]byte(strings.Join([]string{run.ChannelInstanceID.String(), run.HistoryKey, extracted.Type, extracted.Summary}, "\x00"))) + itemID := hex.EncodeToString(hash[:]) + return &store.ChannelMemoryExtractionItem{ + RunID: run.ID, + ChannelInstanceID: run.ChannelInstanceID, + AgentID: run.AgentID, + UserID: run.UserID, + ItemHash: itemID, + ItemType: extracted.Type, + Summary: extracted.Summary, + Topics: topics, + Entities: entities, + Confidence: extracted.Confidence, + SourceID: "channel:" + itemID, + } +} + +func eligibleHistoryKey(key string, cfg Config) bool { + if !cfg.GroupOnly { + return key != "" + } + k := strings.ToLower(key) + return key != "" && !strings.Contains(k, "dm") && !strings.Contains(k, "private") +} + +func messageSourceID(msg store.PendingMessage) string { + if msg.PlatformMsgID != "" { + return msg.PlatformMsgID + } + return msg.ID.String() +} + +func decodeStrings(raw json.RawMessage) []string { + var out []string + _ = json.Unmarshal(raw, &out) + return out +} + +//go:fix inline +func timePtr(t time.Time) *time.Time { return new(t) } + +func contains(values []string, v string) bool { + return slices.Contains(values, v) +} diff --git a/internal/channelmemory/worker.go b/internal/channelmemory/worker.go new file mode 100644 index 00000000..9b1f0cd8 --- /dev/null +++ b/internal/channelmemory/worker.go @@ -0,0 +1,56 @@ +package channelmemory + +import ( + "context" + "log/slog" + "time" + + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +type Worker struct { + Service *Service + Ticker time.Duration +} + +func (w *Worker) Start(ctx context.Context) func() { + if w == nil || w.Service == nil || w.Service.Channels == nil { + return func() {} + } + interval := w.Ticker + if interval <= 0 { + interval = 5 * time.Minute + } + runCtx, cancel := context.WithCancel(ctx) + go func() { + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ticker.C: + w.scan(runCtx) + case <-runCtx.Done(): + return + } + } + }() + return cancel +} + +func (w *Worker) scan(ctx context.Context) { + instances, err := w.Service.Channels.ListAllEnabled(store.WithCrossTenant(ctx)) + if err != nil { + slog.Warn("channel_memory.scan.instances", "error", err) + return + } + for _, inst := range instances { + cfg := ParseConfig(inst.Config) + if !cfg.Enabled { + continue + } + scoped := store.WithTenantID(ctx, inst.TenantID) + if _, err := w.Service.RunNow(scoped, &inst, "scheduled"); err != nil { + slog.Debug("channel_memory.scan.skip", "channel", inst.Name, "error", err) + } + } +} diff --git a/internal/http/channel_instances.go b/internal/http/channel_instances.go index 1f63429f..afe8916f 100644 --- a/internal/http/channel_instances.go +++ b/internal/http/channel_instances.go @@ -11,6 +11,7 @@ import ( "github.com/google/uuid" "github.com/nextlevelbuilder/goclaw/internal/bus" + "github.com/nextlevelbuilder/goclaw/internal/channelmemory" "github.com/nextlevelbuilder/goclaw/internal/channels" "github.com/nextlevelbuilder/goclaw/internal/i18n" "github.com/nextlevelbuilder/goclaw/internal/permissions" @@ -39,6 +40,7 @@ type ChannelInstancesHandler struct { msgBus *bus.MessageBus memberResolver channels.MemberResolver // optional — enriches file_writer metadata on addwriter channelMgr *channels.Manager // optional — enables ChannelDestroyer hook on delete + memoryService *channelmemory.Service mcpStore store.MCPServerStore secureCLIStore store.SecureCLIStore mcpContextStore store.MCPContextAdminStore @@ -144,6 +146,20 @@ func (h *ChannelInstancesHandler) RegisterRoutes(mux *http.ServeMux) { mux.HandleFunc("GET /v1/channels/instances/{id}/contexts/{scopeType}/{scopeKey}/cli-credentials", h.adminAuth(h.handleListContextCLICredentials)) mux.HandleFunc("PUT /v1/channels/instances/{id}/contexts/{scopeType}/{scopeKey}/cli-credentials/{binaryID}", h.adminAuth(h.handleSetContextCLICredentials)) mux.HandleFunc("DELETE /v1/channels/instances/{id}/contexts/{scopeType}/{scopeKey}/cli-credentials/{binaryID}", h.adminAuth(h.handleDeleteContextCLICredentials)) + + if h.memoryService != nil { + mux.HandleFunc("GET /v1/channels/instances/{id}/memory-extraction", h.auth(h.handleMemoryExtractionStatus)) + mux.HandleFunc("PUT /v1/channels/instances/{id}/memory-extraction/settings", h.adminAuth(h.handleMemoryExtractionSettings)) + mux.HandleFunc("POST /v1/channels/instances/{id}/memory-extraction/run", h.adminAuth(h.handleMemoryExtractionRun)) + mux.HandleFunc("GET /v1/channels/instances/{id}/memory-extraction/items", h.auth(h.handleMemoryExtractionItems)) + mux.HandleFunc("POST /v1/channels/instances/{id}/memory-extraction/items/{itemID}/approve", h.adminAuth(h.handleMemoryExtractionApprove)) + mux.HandleFunc("POST /v1/channels/instances/{id}/memory-extraction/items/{itemID}/reject", h.adminAuth(h.handleMemoryExtractionReject)) + mux.HandleFunc("DELETE /v1/channels/instances/{id}/memory-extraction/items/{itemID}", h.adminAuth(h.handleMemoryExtractionDelete)) + } +} + +func (h *ChannelInstancesHandler) SetMemoryExtractionService(svc *channelmemory.Service) { + h.memoryService = svc } func (h *ChannelInstancesHandler) auth(next http.HandlerFunc) http.HandlerFunc { diff --git a/internal/http/channel_memory_extraction.go b/internal/http/channel_memory_extraction.go new file mode 100644 index 00000000..9bb796c2 --- /dev/null +++ b/internal/http/channel_memory_extraction.go @@ -0,0 +1,187 @@ +package http + +import ( + "encoding/json" + "net/http" + + "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/channelmemory" + "github.com/nextlevelbuilder/goclaw/internal/i18n" + "github.com/nextlevelbuilder/goclaw/internal/store" + "github.com/nextlevelbuilder/goclaw/pkg/protocol" +) + +func (h *ChannelInstancesHandler) handleMemoryExtractionStatus(w http.ResponseWriter, r *http.Request) { + inst := h.memoryInstance(w, r) + if inst == nil { + return + } + status, err := h.memoryService.Status(r.Context(), inst) + if err != nil { + writeError(w, http.StatusInternalServerError, protocol.ErrInternal, "failed to load memory extraction status") + return + } + writeJSON(w, http.StatusOK, status) +} + +func (h *ChannelInstancesHandler) handleMemoryExtractionItems(w http.ResponseWriter, r *http.Request) { + inst := h.memoryInstance(w, r) + if inst == nil { + return + } + items, err := h.memoryService.Extractions.ListItems(r.Context(), store.ChannelMemoryItemListOptions{ + ChannelInstanceID: inst.ID, + Status: r.URL.Query().Get("status"), + Limit: 100, + }) + if err != nil { + writeError(w, http.StatusInternalServerError, protocol.ErrInternal, "failed to list memory extraction items") + return + } + writeJSON(w, http.StatusOK, map[string]any{"items": items}) +} + +func (h *ChannelInstancesHandler) handleMemoryExtractionSettings(w http.ResponseWriter, r *http.Request) { + if !requireTenantAdmin(w, r, h.tenantStore) { + return + } + inst := h.memoryInstance(w, r) + if inst == nil { + return + } + var cfg channelmemory.Config + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20)).Decode(&cfg); err != nil { + locale := store.LocaleFromContext(r.Context()) + writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidJSON)) + return + } + normalized := channelmemory.ParseConfig(channelmemory.MergeIntoInstanceConfig(nil, cfg)) + configJSON := channelmemory.MergeIntoInstanceConfig(inst.Config, normalized) + if err := h.store.Update(r.Context(), inst.ID, map[string]any{"config": configJSON}); err != nil { + writeError(w, http.StatusInternalServerError, protocol.ErrInternal, "failed to update memory extraction settings") + return + } + h.emitCacheInvalidate(inst.ID.String()) + emitAudit(h.msgBus, r, "channel_memory.settings_updated", "channel_instance", inst.ID.String()) + writeJSON(w, http.StatusOK, map[string]any{"config": normalized}) +} + +func (h *ChannelInstancesHandler) handleMemoryExtractionRun(w http.ResponseWriter, r *http.Request) { + if !requireTenantAdmin(w, r, h.tenantStore) { + return + } + inst := h.memoryInstance(w, r) + if inst == nil { + return + } + run, err := h.memoryService.RunNow(r.Context(), inst, "manual") + if err != nil { + writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, err.Error()) + return + } + emitAudit(h.msgBus, r, "channel_memory.run_triggered", "channel_instance", inst.ID.String()) + writeJSON(w, http.StatusAccepted, map[string]any{"run": run}) +} + +func (h *ChannelInstancesHandler) handleMemoryExtractionApprove(w http.ResponseWriter, r *http.Request) { + if !requireTenantAdmin(w, r, h.tenantStore) { + return + } + inst := h.memoryInstance(w, r) + if inst == nil { + return + } + itemID, ok := parsePathUUID(w, r, "itemID") + if !ok { + return + } + if !h.memoryItemBelongsToInstance(w, r, itemID, inst.ID) { + return + } + item, err := h.memoryService.Approve(r.Context(), itemID, store.UserIDFromContext(r.Context())) + if err != nil { + writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, err.Error()) + return + } + emitAudit(h.msgBus, r, "channel_memory.item_approved", "channel_memory_item", itemID.String()) + writeJSON(w, http.StatusOK, item) +} + +func (h *ChannelInstancesHandler) handleMemoryExtractionReject(w http.ResponseWriter, r *http.Request) { + if !requireTenantAdmin(w, r, h.tenantStore) { + return + } + inst := h.memoryInstance(w, r) + if inst == nil { + return + } + itemID, ok := parsePathUUID(w, r, "itemID") + if !ok { + return + } + if !h.memoryItemBelongsToInstance(w, r, itemID, inst.ID) { + return + } + if err := h.memoryService.Reject(r.Context(), itemID, store.UserIDFromContext(r.Context())); err != nil { + writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, err.Error()) + return + } + emitAudit(h.msgBus, r, "channel_memory.item_rejected", "channel_memory_item", itemID.String()) + writeJSON(w, http.StatusOK, map[string]string{"status": "rejected"}) +} + +func (h *ChannelInstancesHandler) handleMemoryExtractionDelete(w http.ResponseWriter, r *http.Request) { + if !requireTenantAdmin(w, r, h.tenantStore) { + return + } + inst := h.memoryInstance(w, r) + if inst == nil { + return + } + itemID, ok := parsePathUUID(w, r, "itemID") + if !ok { + return + } + if !h.memoryItemBelongsToInstance(w, r, itemID, inst.ID) { + return + } + if err := h.memoryService.Delete(r.Context(), itemID); err != nil { + writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, err.Error()) + return + } + emitAudit(h.msgBus, r, "channel_memory.item_deleted", "channel_memory_item", itemID.String()) + writeJSON(w, http.StatusOK, map[string]string{"status": "deleted"}) +} + +func (h *ChannelInstancesHandler) memoryInstance(w http.ResponseWriter, r *http.Request) *store.ChannelInstanceData { + id, ok := parsePathUUID(w, r, "id") + if !ok { + return nil + } + inst, err := h.store.Get(r.Context(), id) + if err != nil { + locale := store.LocaleFromContext(r.Context()) + writeError(w, http.StatusNotFound, protocol.ErrNotFound, i18n.T(locale, i18n.MsgInstanceNotFound)) + return nil + } + return inst +} + +func parsePathUUID(w http.ResponseWriter, r *http.Request, name string) (uuid.UUID, bool) { + id, err := uuid.Parse(r.PathValue(name)) + if err != nil { + locale := store.LocaleFromContext(r.Context()) + writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidID, name)) + return uuid.Nil, false + } + return id, true +} + +func (h *ChannelInstancesHandler) memoryItemBelongsToInstance(w http.ResponseWriter, r *http.Request, itemID, channelID uuid.UUID) bool { + item, err := h.memoryService.Extractions.GetItem(r.Context(), itemID) + if err != nil || item.ChannelInstanceID != channelID { + writeError(w, http.StatusNotFound, protocol.ErrNotFound, "memory extraction item not found") + return false + } + return true +} diff --git a/internal/store/channel_memory_extraction_store.go b/internal/store/channel_memory_extraction_store.go new file mode 100644 index 00000000..d6f70c4b --- /dev/null +++ b/internal/store/channel_memory_extraction_store.go @@ -0,0 +1,99 @@ +package store + +import ( + "context" + "encoding/json" + "time" + + "github.com/google/uuid" +) + +const ( + ChannelMemoryRunPending = "pending" + ChannelMemoryRunRunning = "running" + ChannelMemoryRunCompleted = "completed" + ChannelMemoryRunFailed = "failed" + + ChannelMemoryItemPendingReview = "pending_review" + ChannelMemoryItemApproved = "approved" + ChannelMemoryItemRejected = "rejected" + ChannelMemoryItemWritten = "written" + ChannelMemoryItemDeleted = "deleted" +) + +type ChannelMemoryExtractionRun struct { + ID uuid.UUID `json:"id" db:"id"` + TenantID uuid.UUID `json:"tenant_id" db:"tenant_id"` + ChannelInstanceID uuid.UUID `json:"channel_instance_id" db:"channel_instance_id"` + ChannelName string `json:"channel_name" db:"channel_name"` + AgentID uuid.UUID `json:"agent_id" db:"agent_id"` + UserID string `json:"user_id" db:"user_id"` + HistoryKey string `json:"history_key" db:"history_key"` + Trigger string `json:"trigger" db:"trigger"` + Status string `json:"status" db:"status"` + SourceStartID string `json:"source_start_id" db:"source_start_id"` + SourceEndID string `json:"source_end_id" db:"source_end_id"` + SourceStartAt *time.Time `json:"source_start_at,omitempty" db:"source_start_at"` + SourceEndAt *time.Time `json:"source_end_at,omitempty" db:"source_end_at"` + MessageCount int `json:"message_count" db:"message_count"` + RedactionCount int `json:"redaction_count" db:"redaction_count"` + RedactionTypes json.RawMessage `json:"redaction_types" db:"redaction_types"` + ItemCount int `json:"item_count" db:"item_count"` + ErrorMessage string `json:"error_message,omitempty" db:"error_message"` + StartedAt *time.Time `json:"started_at,omitempty" db:"started_at"` + CompletedAt *time.Time `json:"completed_at,omitempty" db:"completed_at"` + CreatedAt time.Time `json:"created_at" db:"created_at"` + UpdatedAt time.Time `json:"updated_at" db:"updated_at"` +} + +type ChannelMemoryExtractionItem struct { + ID uuid.UUID `json:"id" db:"id"` + TenantID uuid.UUID `json:"tenant_id" db:"tenant_id"` + RunID uuid.UUID `json:"run_id" db:"run_id"` + ChannelInstanceID uuid.UUID `json:"channel_instance_id" db:"channel_instance_id"` + AgentID uuid.UUID `json:"agent_id" db:"agent_id"` + UserID string `json:"user_id" db:"user_id"` + ItemHash string `json:"item_hash" db:"item_hash"` + ItemType string `json:"item_type" db:"item_type"` + Summary string `json:"summary" db:"summary"` + Topics json.RawMessage `json:"topics" db:"topics"` + Entities json.RawMessage `json:"entities" db:"entities"` + Confidence float64 `json:"confidence" db:"confidence"` + SourceID string `json:"source_id" db:"source_id"` + Status string `json:"status" db:"status"` + ApprovedBy string `json:"approved_by,omitempty" db:"approved_by"` + ApprovedAt *time.Time `json:"approved_at,omitempty" db:"approved_at"` + RejectedBy string `json:"rejected_by,omitempty" db:"rejected_by"` + RejectedAt *time.Time `json:"rejected_at,omitempty" db:"rejected_at"` + DeletedAt *time.Time `json:"deleted_at,omitempty" db:"deleted_at"` + WrittenAt *time.Time `json:"written_at,omitempty" db:"written_at"` + EpisodicID string `json:"episodic_id,omitempty" db:"episodic_id"` + CreatedAt time.Time `json:"created_at" db:"created_at"` + UpdatedAt time.Time `json:"updated_at" db:"updated_at"` +} + +type ChannelMemoryRunListOptions struct { + ChannelInstanceID uuid.UUID + HistoryKey string + Status string + Limit int +} + +type ChannelMemoryItemListOptions struct { + ChannelInstanceID uuid.UUID + RunID uuid.UUID + Status string + Limit int +} + +type ChannelMemoryExtractionStore interface { + CreateRun(ctx context.Context, run *ChannelMemoryExtractionRun) error + GetRun(ctx context.Context, id uuid.UUID) (*ChannelMemoryExtractionRun, error) + ListRuns(ctx context.Context, opts ChannelMemoryRunListOptions) ([]ChannelMemoryExtractionRun, error) + UpdateRun(ctx context.Context, id uuid.UUID, updates map[string]any) error + + CreateItem(ctx context.Context, item *ChannelMemoryExtractionItem) error + GetItem(ctx context.Context, id uuid.UUID) (*ChannelMemoryExtractionItem, error) + ListItems(ctx context.Context, opts ChannelMemoryItemListOptions) ([]ChannelMemoryExtractionItem, error) + UpdateItem(ctx context.Context, id uuid.UUID, updates map[string]any) error +} diff --git a/internal/store/pg/channel_memory_extraction.go b/internal/store/pg/channel_memory_extraction.go new file mode 100644 index 00000000..9a802855 --- /dev/null +++ b/internal/store/pg/channel_memory_extraction.go @@ -0,0 +1,239 @@ +package pg + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "strings" + "time" + + "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +type PGChannelMemoryExtractionStore struct { + db *sql.DB +} + +func NewPGChannelMemoryExtractionStore(db *sql.DB) *PGChannelMemoryExtractionStore { + return &PGChannelMemoryExtractionStore{db: db} +} + +const channelMemoryRunCols = `id, tenant_id, channel_instance_id, channel_name, agent_id, user_id, + history_key, trigger, status, source_start_id, source_end_id, source_start_at, source_end_at, + message_count, redaction_count, redaction_types, item_count, error_message, started_at, + completed_at, created_at, updated_at` + +const channelMemoryItemCols = `id, tenant_id, run_id, channel_instance_id, agent_id, user_id, + item_hash, item_type, summary, topics, entities, confidence, source_id, status, approved_by, + approved_at, rejected_by, rejected_at, deleted_at, written_at, episodic_id, created_at, updated_at` + +func (s *PGChannelMemoryExtractionStore) CreateRun(ctx context.Context, run *store.ChannelMemoryExtractionRun) error { + if run.ID == uuid.Nil { + run.ID = store.GenNewID() + } + run.TenantID = tenantIDForInsert(ctx) + now := time.Now().UTC() + run.CreatedAt = now + run.UpdatedAt = now + if run.Status == "" { + run.Status = store.ChannelMemoryRunPending + } + if run.RedactionTypes == nil { + run.RedactionTypes = json.RawMessage("[]") + } + err := s.db.QueryRowContext(ctx, `INSERT INTO channel_memory_extraction_runs + (`+channelMemoryRunCols+`) + 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, channel_instance_id, history_key, source_start_id, source_end_id) + DO UPDATE SET updated_at = EXCLUDED.updated_at + RETURNING id`, + run.ID, run.TenantID, run.ChannelInstanceID, run.ChannelName, run.AgentID, run.UserID, + run.HistoryKey, run.Trigger, run.Status, run.SourceStartID, run.SourceEndID, + run.SourceStartAt, run.SourceEndAt, run.MessageCount, run.RedactionCount, + run.RedactionTypes, run.ItemCount, run.ErrorMessage, run.StartedAt, run.CompletedAt, + run.CreatedAt, run.UpdatedAt, + ).Scan(&run.ID) + return err +} + +func (s *PGChannelMemoryExtractionStore) GetRun(ctx context.Context, id uuid.UUID) (*store.ChannelMemoryExtractionRun, error) { + tClause, tArgs, _, err := scopeClause(ctx, 2) + if err != nil { + return nil, err + } + row := s.db.QueryRowContext(ctx, `SELECT `+channelMemoryRunCols+` FROM channel_memory_extraction_runs WHERE id = $1`+tClause, append([]any{id}, tArgs...)...) + return scanChannelMemoryRun(row) +} + +func (s *PGChannelMemoryExtractionStore) ListRuns(ctx context.Context, opts store.ChannelMemoryRunListOptions) ([]store.ChannelMemoryExtractionRun, error) { + where, args, next, err := scopeClause(ctx, 1) + if err != nil { + return nil, err + } + conds := []string{"1=1" + where} + if opts.ChannelInstanceID != uuid.Nil { + conds = append(conds, fmt.Sprintf("channel_instance_id = $%d", next)) + args = append(args, opts.ChannelInstanceID) + next++ + } + if opts.HistoryKey != "" { + conds = append(conds, fmt.Sprintf("history_key = $%d", next)) + args = append(args, opts.HistoryKey) + next++ + } + if opts.Status != "" { + conds = append(conds, fmt.Sprintf("status = $%d", next)) + args = append(args, opts.Status) + next++ + } + limit := opts.Limit + if limit <= 0 || limit > 200 { + limit = 50 + } + args = append(args, limit) + rows, err := s.db.QueryContext(ctx, `SELECT `+channelMemoryRunCols+` + FROM channel_memory_extraction_runs WHERE `+strings.Join(conds, " AND ")+` + ORDER BY created_at DESC LIMIT $`+fmt.Sprint(next), args...) + if err != nil { + return nil, err + } + defer rows.Close() + var out []store.ChannelMemoryExtractionRun + for rows.Next() { + run, err := scanChannelMemoryRunRows(rows) + if err != nil { + return nil, err + } + out = append(out, *run) + } + return out, rows.Err() +} + +func (s *PGChannelMemoryExtractionStore) UpdateRun(ctx context.Context, id uuid.UUID, updates map[string]any) error { + tid, err := requireTenantID(ctx) + if err != nil { + return err + } + return execMapUpdateWhereTenant(ctx, s.db, "channel_memory_extraction_runs", updates, id, tid) +} + +func (s *PGChannelMemoryExtractionStore) CreateItem(ctx context.Context, item *store.ChannelMemoryExtractionItem) error { + if item.ID == uuid.Nil { + item.ID = store.GenNewID() + } + item.TenantID = tenantIDForInsert(ctx) + now := time.Now().UTC() + item.CreatedAt = now + item.UpdatedAt = now + if item.Status == "" { + item.Status = store.ChannelMemoryItemPendingReview + } + if item.Topics == nil { + item.Topics = json.RawMessage("[]") + } + if item.Entities == nil { + item.Entities = json.RawMessage("[]") + } + err := s.db.QueryRowContext(ctx, `INSERT INTO channel_memory_extraction_items + (`+channelMemoryItemCols+`) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23) + ON CONFLICT (tenant_id, run_id, item_hash) DO UPDATE SET updated_at = EXCLUDED.updated_at + RETURNING id`, + item.ID, item.TenantID, item.RunID, item.ChannelInstanceID, item.AgentID, item.UserID, + item.ItemHash, item.ItemType, item.Summary, item.Topics, item.Entities, item.Confidence, + item.SourceID, item.Status, item.ApprovedBy, item.ApprovedAt, item.RejectedBy, + item.RejectedAt, item.DeletedAt, item.WrittenAt, item.EpisodicID, item.CreatedAt, item.UpdatedAt, + ).Scan(&item.ID) + return err +} + +func (s *PGChannelMemoryExtractionStore) GetItem(ctx context.Context, id uuid.UUID) (*store.ChannelMemoryExtractionItem, error) { + tClause, tArgs, _, err := scopeClause(ctx, 2) + if err != nil { + return nil, err + } + row := s.db.QueryRowContext(ctx, `SELECT `+channelMemoryItemCols+` FROM channel_memory_extraction_items WHERE id = $1`+tClause, append([]any{id}, tArgs...)...) + return scanChannelMemoryItem(row) +} + +func (s *PGChannelMemoryExtractionStore) ListItems(ctx context.Context, opts store.ChannelMemoryItemListOptions) ([]store.ChannelMemoryExtractionItem, error) { + where, args, next, err := scopeClause(ctx, 1) + if err != nil { + return nil, err + } + conds := []string{"1=1" + where} + if opts.ChannelInstanceID != uuid.Nil { + conds = append(conds, fmt.Sprintf("channel_instance_id = $%d", next)) + args = append(args, opts.ChannelInstanceID) + next++ + } + if opts.RunID != uuid.Nil { + conds = append(conds, fmt.Sprintf("run_id = $%d", next)) + args = append(args, opts.RunID) + next++ + } + if opts.Status != "" { + conds = append(conds, fmt.Sprintf("status = $%d", next)) + args = append(args, opts.Status) + next++ + } + limit := opts.Limit + if limit <= 0 || limit > 200 { + limit = 50 + } + args = append(args, limit) + rows, err := s.db.QueryContext(ctx, `SELECT `+channelMemoryItemCols+` + FROM channel_memory_extraction_items WHERE `+strings.Join(conds, " AND ")+` + ORDER BY created_at DESC LIMIT $`+fmt.Sprint(next), args...) + if err != nil { + return nil, err + } + defer rows.Close() + var out []store.ChannelMemoryExtractionItem + for rows.Next() { + item, err := scanChannelMemoryItemRows(rows) + if err != nil { + return nil, err + } + out = append(out, *item) + } + return out, rows.Err() +} + +func (s *PGChannelMemoryExtractionStore) UpdateItem(ctx context.Context, id uuid.UUID, updates map[string]any) error { + tid, err := requireTenantID(ctx) + if err != nil { + return err + } + return execMapUpdateWhereTenant(ctx, s.db, "channel_memory_extraction_items", updates, id, tid) +} + +type channelMemoryScanner interface { + Scan(dest ...any) error +} + +func scanChannelMemoryRun(row channelMemoryScanner) (*store.ChannelMemoryExtractionRun, error) { + var r store.ChannelMemoryExtractionRun + if err := row.Scan(&r.ID, &r.TenantID, &r.ChannelInstanceID, &r.ChannelName, &r.AgentID, &r.UserID, &r.HistoryKey, &r.Trigger, &r.Status, &r.SourceStartID, &r.SourceEndID, &r.SourceStartAt, &r.SourceEndAt, &r.MessageCount, &r.RedactionCount, &r.RedactionTypes, &r.ItemCount, &r.ErrorMessage, &r.StartedAt, &r.CompletedAt, &r.CreatedAt, &r.UpdatedAt); err != nil { + return nil, err + } + return &r, nil +} + +func scanChannelMemoryRunRows(rows *sql.Rows) (*store.ChannelMemoryExtractionRun, error) { + return scanChannelMemoryRun(rows) +} + +func scanChannelMemoryItem(row channelMemoryScanner) (*store.ChannelMemoryExtractionItem, error) { + var i store.ChannelMemoryExtractionItem + if err := row.Scan(&i.ID, &i.TenantID, &i.RunID, &i.ChannelInstanceID, &i.AgentID, &i.UserID, &i.ItemHash, &i.ItemType, &i.Summary, &i.Topics, &i.Entities, &i.Confidence, &i.SourceID, &i.Status, &i.ApprovedBy, &i.ApprovedAt, &i.RejectedBy, &i.RejectedAt, &i.DeletedAt, &i.WrittenAt, &i.EpisodicID, &i.CreatedAt, &i.UpdatedAt); err != nil { + return nil, err + } + return &i, nil +} + +func scanChannelMemoryItemRows(rows *sql.Rows) (*store.ChannelMemoryExtractionItem, error) { + return scanChannelMemoryItem(rows) +} diff --git a/internal/store/pg/factory.go b/internal/store/pg/factory.go index 6cee2461..ea2bc565 100644 --- a/internal/store/pg/factory.go +++ b/internal/store/pg/factory.go @@ -41,6 +41,7 @@ func NewPGStores(cfg store.StoreConfig) (*store.Stores, error) { Teams: NewPGTeamStore(db), BuiltinTools: NewPGBuiltinToolStore(db), PendingMessages: NewPGPendingMessageStore(db), + ChannelMemory: NewPGChannelMemoryExtractionStore(db), KnowledgeGraph: NewPGKnowledgeGraphStore(db), Contacts: NewPGContactStore(db), Activity: NewPGActivityStore(db), diff --git a/internal/store/sqlitestore/channel-memory-extraction.go b/internal/store/sqlitestore/channel-memory-extraction.go new file mode 100644 index 00000000..ff72def7 --- /dev/null +++ b/internal/store/sqlitestore/channel-memory-extraction.go @@ -0,0 +1,257 @@ +//go:build sqlite || sqliteonly + +package sqlitestore + +import ( + "context" + "database/sql" + "encoding/json" + "strings" + "time" + + "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +type SQLiteChannelMemoryExtractionStore struct { + db *sql.DB +} + +func NewSQLiteChannelMemoryExtractionStore(db *sql.DB) *SQLiteChannelMemoryExtractionStore { + return &SQLiteChannelMemoryExtractionStore{db: db} +} + +const sqliteChannelMemoryRunCols = `id, tenant_id, channel_instance_id, channel_name, agent_id, user_id, + history_key, trigger, status, source_start_id, source_end_id, source_start_at, source_end_at, + message_count, redaction_count, redaction_types, item_count, error_message, started_at, + completed_at, created_at, updated_at` + +const sqliteChannelMemoryItemCols = `id, tenant_id, run_id, channel_instance_id, agent_id, user_id, + item_hash, item_type, summary, topics, entities, confidence, source_id, status, approved_by, + approved_at, rejected_by, rejected_at, deleted_at, written_at, episodic_id, created_at, updated_at` + +func (s *SQLiteChannelMemoryExtractionStore) CreateRun(ctx context.Context, run *store.ChannelMemoryExtractionRun) error { + if run.ID == uuid.Nil { + run.ID = store.GenNewID() + } + run.TenantID = tenantIDForInsert(ctx) + now := time.Now().UTC() + run.CreatedAt = now + run.UpdatedAt = now + if run.Status == "" { + run.Status = store.ChannelMemoryRunPending + } + if run.RedactionTypes == nil { + run.RedactionTypes = json.RawMessage("[]") + } + return s.db.QueryRowContext(ctx, `INSERT INTO channel_memory_extraction_runs + (`+sqliteChannelMemoryRunCols+`) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) + ON CONFLICT (tenant_id, channel_instance_id, history_key, source_start_id, source_end_id) + DO UPDATE SET updated_at = excluded.updated_at + RETURNING id`, + run.ID, run.TenantID, run.ChannelInstanceID, run.ChannelName, run.AgentID, run.UserID, + run.HistoryKey, run.Trigger, run.Status, run.SourceStartID, run.SourceEndID, + run.SourceStartAt, run.SourceEndAt, run.MessageCount, run.RedactionCount, + string(run.RedactionTypes), run.ItemCount, run.ErrorMessage, run.StartedAt, run.CompletedAt, + run.CreatedAt, run.UpdatedAt, + ).Scan(&run.ID) +} + +func (s *SQLiteChannelMemoryExtractionStore) GetRun(ctx context.Context, id uuid.UUID) (*store.ChannelMemoryExtractionRun, error) { + tClause, tArgs, err := scopeClause(ctx) + if err != nil { + return nil, err + } + row := s.db.QueryRowContext(ctx, `SELECT `+sqliteChannelMemoryRunCols+` FROM channel_memory_extraction_runs WHERE id = ?`+tClause, append([]any{id}, tArgs...)...) + return scanSQLiteChannelMemoryRun(row) +} + +func (s *SQLiteChannelMemoryExtractionStore) ListRuns(ctx context.Context, opts store.ChannelMemoryRunListOptions) ([]store.ChannelMemoryExtractionRun, error) { + tClause, args, err := scopeClause(ctx) + if err != nil { + return nil, err + } + conds := []string{"1=1" + tClause} + if opts.ChannelInstanceID != uuid.Nil { + conds = append(conds, "channel_instance_id = ?") + args = append(args, opts.ChannelInstanceID) + } + if opts.HistoryKey != "" { + conds = append(conds, "history_key = ?") + args = append(args, opts.HistoryKey) + } + if opts.Status != "" { + conds = append(conds, "status = ?") + args = append(args, opts.Status) + } + limit := opts.Limit + if limit <= 0 || limit > 200 { + limit = 50 + } + args = append(args, limit) + rows, err := s.db.QueryContext(ctx, `SELECT `+sqliteChannelMemoryRunCols+` + FROM channel_memory_extraction_runs WHERE `+strings.Join(conds, " AND ")+` + ORDER BY created_at DESC LIMIT ?`, args...) + if err != nil { + return nil, err + } + defer rows.Close() + var out []store.ChannelMemoryExtractionRun + for rows.Next() { + run, err := scanSQLiteChannelMemoryRun(rows) + if err != nil { + return nil, err + } + out = append(out, *run) + } + return out, rows.Err() +} + +func (s *SQLiteChannelMemoryExtractionStore) UpdateRun(ctx context.Context, id uuid.UUID, updates map[string]any) error { + tid, err := requireTenantID(ctx) + if err != nil { + return err + } + return execMapUpdateWhereTenant(ctx, s.db, "channel_memory_extraction_runs", updates, id, tid) +} + +func (s *SQLiteChannelMemoryExtractionStore) CreateItem(ctx context.Context, item *store.ChannelMemoryExtractionItem) error { + if item.ID == uuid.Nil { + item.ID = store.GenNewID() + } + item.TenantID = tenantIDForInsert(ctx) + now := time.Now().UTC() + item.CreatedAt = now + item.UpdatedAt = now + if item.Status == "" { + item.Status = store.ChannelMemoryItemPendingReview + } + if item.Topics == nil { + item.Topics = json.RawMessage("[]") + } + if item.Entities == nil { + item.Entities = json.RawMessage("[]") + } + return s.db.QueryRowContext(ctx, `INSERT INTO channel_memory_extraction_items + (`+sqliteChannelMemoryItemCols+`) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) + ON CONFLICT (tenant_id, run_id, item_hash) DO UPDATE SET updated_at = excluded.updated_at + RETURNING id`, + item.ID, item.TenantID, item.RunID, item.ChannelInstanceID, item.AgentID, item.UserID, + item.ItemHash, item.ItemType, item.Summary, string(item.Topics), string(item.Entities), + item.Confidence, item.SourceID, item.Status, item.ApprovedBy, item.ApprovedAt, + item.RejectedBy, item.RejectedAt, item.DeletedAt, item.WrittenAt, item.EpisodicID, + item.CreatedAt, item.UpdatedAt, + ).Scan(&item.ID) +} + +func (s *SQLiteChannelMemoryExtractionStore) GetItem(ctx context.Context, id uuid.UUID) (*store.ChannelMemoryExtractionItem, error) { + tClause, tArgs, err := scopeClause(ctx) + if err != nil { + return nil, err + } + row := s.db.QueryRowContext(ctx, `SELECT `+sqliteChannelMemoryItemCols+` FROM channel_memory_extraction_items WHERE id = ?`+tClause, append([]any{id}, tArgs...)...) + return scanSQLiteChannelMemoryItem(row) +} + +func (s *SQLiteChannelMemoryExtractionStore) ListItems(ctx context.Context, opts store.ChannelMemoryItemListOptions) ([]store.ChannelMemoryExtractionItem, error) { + tClause, args, err := scopeClause(ctx) + if err != nil { + return nil, err + } + conds := []string{"1=1" + tClause} + if opts.ChannelInstanceID != uuid.Nil { + conds = append(conds, "channel_instance_id = ?") + args = append(args, opts.ChannelInstanceID) + } + if opts.RunID != uuid.Nil { + conds = append(conds, "run_id = ?") + args = append(args, opts.RunID) + } + if opts.Status != "" { + conds = append(conds, "status = ?") + args = append(args, opts.Status) + } + limit := opts.Limit + if limit <= 0 || limit > 200 { + limit = 50 + } + args = append(args, limit) + rows, err := s.db.QueryContext(ctx, `SELECT `+sqliteChannelMemoryItemCols+` + FROM channel_memory_extraction_items WHERE `+strings.Join(conds, " AND ")+` + ORDER BY created_at DESC LIMIT ?`, args...) + if err != nil { + return nil, err + } + defer rows.Close() + var out []store.ChannelMemoryExtractionItem + for rows.Next() { + item, err := scanSQLiteChannelMemoryItem(rows) + if err != nil { + return nil, err + } + out = append(out, *item) + } + return out, rows.Err() +} + +func (s *SQLiteChannelMemoryExtractionStore) UpdateItem(ctx context.Context, id uuid.UUID, updates map[string]any) error { + tid, err := requireTenantID(ctx) + if err != nil { + return err + } + return execMapUpdateWhereTenant(ctx, s.db, "channel_memory_extraction_items", updates, id, tid) +} + +type sqliteChannelMemoryScanner interface { + Scan(dest ...any) error +} + +func scanSQLiteChannelMemoryRun(row sqliteChannelMemoryScanner) (*store.ChannelMemoryExtractionRun, error) { + var r store.ChannelMemoryExtractionRun + var redactions string + sourceStart, sourceEnd := &nullSqliteTime{}, &nullSqliteTime{} + started, completed := &nullSqliteTime{}, &nullSqliteTime{} + created, updated := scanTimePair() + if err := row.Scan(&r.ID, &r.TenantID, &r.ChannelInstanceID, &r.ChannelName, &r.AgentID, &r.UserID, &r.HistoryKey, &r.Trigger, &r.Status, &r.SourceStartID, &r.SourceEndID, sourceStart, sourceEnd, &r.MessageCount, &r.RedactionCount, &redactions, &r.ItemCount, &r.ErrorMessage, started, completed, created, updated); err != nil { + return nil, err + } + r.RedactionTypes = json.RawMessage(redactions) + r.SourceStartAt = sqliteTimePtr(sourceStart) + r.SourceEndAt = sqliteTimePtr(sourceEnd) + r.StartedAt = sqliteTimePtr(started) + r.CompletedAt = sqliteTimePtr(completed) + r.CreatedAt = created.Time + r.UpdatedAt = updated.Time + return &r, nil +} + +func scanSQLiteChannelMemoryItem(row sqliteChannelMemoryScanner) (*store.ChannelMemoryExtractionItem, error) { + var i store.ChannelMemoryExtractionItem + var topics, entities string + approved, rejected := &nullSqliteTime{}, &nullSqliteTime{} + deleted, written := &nullSqliteTime{}, &nullSqliteTime{} + created, updated := scanTimePair() + if err := row.Scan(&i.ID, &i.TenantID, &i.RunID, &i.ChannelInstanceID, &i.AgentID, &i.UserID, &i.ItemHash, &i.ItemType, &i.Summary, &topics, &entities, &i.Confidence, &i.SourceID, &i.Status, &i.ApprovedBy, approved, &i.RejectedBy, rejected, deleted, written, &i.EpisodicID, created, updated); err != nil { + return nil, err + } + i.Topics = json.RawMessage(topics) + i.Entities = json.RawMessage(entities) + i.ApprovedAt = sqliteTimePtr(approved) + i.RejectedAt = sqliteTimePtr(rejected) + i.DeletedAt = sqliteTimePtr(deleted) + i.WrittenAt = sqliteTimePtr(written) + i.CreatedAt = created.Time + i.UpdatedAt = updated.Time + return &i, nil +} + +func sqliteTimePtr(nt *nullSqliteTime) *time.Time { + if nt == nil || !nt.Valid { + return nil + } + return &nt.Time +} + +var _ store.ChannelMemoryExtractionStore = (*SQLiteChannelMemoryExtractionStore)(nil) diff --git a/internal/store/sqlitestore/factory.go b/internal/store/sqlitestore/factory.go index 296ca92d..284c4c1a 100644 --- a/internal/store/sqlitestore/factory.go +++ b/internal/store/sqlitestore/factory.go @@ -54,6 +54,7 @@ func NewSQLiteStores(cfg store.StoreConfig) (*store.Stores, error) { ChannelInstances: NewSQLiteChannelInstanceStore(db, cfg.EncryptionKey), Pairing: NewSQLitePairingStore(db), PendingMessages: NewSQLitePendingMessageStore(db), + ChannelMemory: NewSQLiteChannelMemoryExtractionStore(db), Contacts: NewSQLiteContactStore(db), Teams: NewSQLiteTeamStore(db), Skills: NewSQLiteSkillStore(db, cfg.SkillsStorageDir), diff --git a/internal/store/sqlitestore/schema.go b/internal/store/sqlitestore/schema.go index b0583def..ac4e0193 100644 --- a/internal/store/sqlitestore/schema.go +++ b/internal/store/sqlitestore/schema.go @@ -16,7 +16,7 @@ var schemaSQL string // SchemaVersion is the current SQLite schema version. // Bump this when adding new migration steps below. -const SchemaVersion = 44 +const SchemaVersion = 45 // migrations maps version → SQL to apply when upgrading FROM that version. // schema.sql always represents the LATEST full schema (for fresh DBs). @@ -805,8 +805,72 @@ CREATE INDEX IF NOT EXISTS idx_run_timeline_trace WHERE trace_id IS NOT NULL;`, // Version 43 → 44: channel-context MCP and Secure CLI grants/credentials. 43: addChannelContextCapabilityTables, + // Version 44 → 45: passive channel memory extraction run and review queue. + 44: addChannelMemoryExtractionTables, } +const addChannelMemoryExtractionTables = ` +CREATE TABLE IF NOT EXISTS channel_memory_extraction_runs ( + id TEXT NOT NULL PRIMARY KEY, + tenant_id TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE, + channel_instance_id TEXT NOT NULL REFERENCES channel_instances(id) ON DELETE CASCADE, + channel_name VARCHAR(255) NOT NULL, + agent_id TEXT NOT NULL REFERENCES agents(id) ON DELETE CASCADE, + user_id VARCHAR(255) NOT NULL DEFAULT '', + history_key VARCHAR(255) NOT NULL, + trigger VARCHAR(32) NOT NULL DEFAULT 'scheduled', + status VARCHAR(32) NOT NULL DEFAULT 'pending', + source_start_id VARCHAR(255) NOT NULL DEFAULT '', + source_end_id VARCHAR(255) NOT NULL DEFAULT '', + source_start_at TEXT, + source_end_at TEXT, + message_count INTEGER NOT NULL DEFAULT 0, + redaction_count INTEGER NOT NULL DEFAULT 0, + redaction_types TEXT NOT NULL DEFAULT '[]', + item_count INTEGER NOT NULL DEFAULT 0, + error_message TEXT NOT NULL DEFAULT '', + started_at TEXT, + completed_at TEXT, + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), + updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), + UNIQUE (tenant_id, channel_instance_id, history_key, source_start_id, source_end_id) +); +CREATE INDEX IF NOT EXISTS idx_channel_memory_runs_channel + ON channel_memory_extraction_runs(tenant_id, channel_instance_id, created_at DESC); +CREATE INDEX IF NOT EXISTS idx_channel_memory_runs_status + ON channel_memory_extraction_runs(tenant_id, status, created_at DESC); + +CREATE TABLE IF NOT EXISTS channel_memory_extraction_items ( + id TEXT NOT NULL PRIMARY KEY, + tenant_id TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE, + run_id TEXT NOT NULL REFERENCES channel_memory_extraction_runs(id) ON DELETE CASCADE, + channel_instance_id TEXT NOT NULL REFERENCES channel_instances(id) ON DELETE CASCADE, + agent_id TEXT NOT NULL REFERENCES agents(id) ON DELETE CASCADE, + user_id VARCHAR(255) NOT NULL DEFAULT '', + item_hash VARCHAR(128) NOT NULL, + item_type VARCHAR(64) NOT NULL, + summary TEXT NOT NULL, + topics TEXT NOT NULL DEFAULT '[]', + entities TEXT NOT NULL DEFAULT '[]', + confidence REAL NOT NULL DEFAULT 0, + source_id VARCHAR(255) NOT NULL DEFAULT '', + status VARCHAR(32) NOT NULL DEFAULT 'pending_review', + approved_by VARCHAR(255) NOT NULL DEFAULT '', + approved_at TEXT, + rejected_by VARCHAR(255) NOT NULL DEFAULT '', + rejected_at TEXT, + deleted_at TEXT, + written_at TEXT, + episodic_id VARCHAR(64) NOT NULL DEFAULT '', + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), + updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), + UNIQUE (tenant_id, run_id, item_hash) +); +CREATE INDEX IF NOT EXISTS idx_channel_memory_items_channel_status + ON channel_memory_extraction_items(tenant_id, channel_instance_id, status, created_at DESC); +CREATE INDEX IF NOT EXISTS idx_channel_memory_items_run + ON channel_memory_extraction_items(tenant_id, run_id);` + const addChannelContextCapabilityTables = ` CREATE TABLE IF NOT EXISTS mcp_context_grants ( id TEXT NOT NULL PRIMARY KEY, diff --git a/internal/store/sqlitestore/schema.sql b/internal/store/sqlitestore/schema.sql index a577d7a8..c46fc0a3 100644 --- a/internal/store/sqlitestore/schema.sql +++ b/internal/store/sqlitestore/schema.sql @@ -1177,6 +1177,77 @@ CREATE TABLE IF NOT EXISTS channel_pending_messages ( CREATE INDEX IF NOT EXISTS idx_channel_pending_messages_lookup ON channel_pending_messages(channel_name, history_key, created_at); CREATE INDEX IF NOT EXISTS idx_channel_pending_messages_tenant ON channel_pending_messages(tenant_id); +-- ============================================================ +-- Table: channel_memory_extraction_runs +-- ============================================================ + +CREATE TABLE IF NOT EXISTS channel_memory_extraction_runs ( + id TEXT NOT NULL PRIMARY KEY, + tenant_id TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE, + channel_instance_id TEXT NOT NULL REFERENCES channel_instances(id) ON DELETE CASCADE, + channel_name VARCHAR(255) NOT NULL, + agent_id TEXT NOT NULL REFERENCES agents(id) ON DELETE CASCADE, + user_id VARCHAR(255) NOT NULL DEFAULT '', + history_key VARCHAR(255) NOT NULL, + trigger VARCHAR(32) NOT NULL DEFAULT 'scheduled', + status VARCHAR(32) NOT NULL DEFAULT 'pending', + source_start_id VARCHAR(255) NOT NULL DEFAULT '', + source_end_id VARCHAR(255) NOT NULL DEFAULT '', + source_start_at TEXT, + source_end_at TEXT, + message_count INTEGER NOT NULL DEFAULT 0, + redaction_count INTEGER NOT NULL DEFAULT 0, + redaction_types TEXT NOT NULL DEFAULT '[]', + item_count INTEGER NOT NULL DEFAULT 0, + error_message TEXT NOT NULL DEFAULT '', + started_at TEXT, + completed_at TEXT, + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), + updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), + UNIQUE (tenant_id, channel_instance_id, history_key, source_start_id, source_end_id) +); + +CREATE INDEX IF NOT EXISTS idx_channel_memory_runs_channel + ON channel_memory_extraction_runs(tenant_id, channel_instance_id, created_at DESC); +CREATE INDEX IF NOT EXISTS idx_channel_memory_runs_status + ON channel_memory_extraction_runs(tenant_id, status, created_at DESC); + +-- ============================================================ +-- Table: channel_memory_extraction_items +-- ============================================================ + +CREATE TABLE IF NOT EXISTS channel_memory_extraction_items ( + id TEXT NOT NULL PRIMARY KEY, + tenant_id TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE, + run_id TEXT NOT NULL REFERENCES channel_memory_extraction_runs(id) ON DELETE CASCADE, + channel_instance_id TEXT NOT NULL REFERENCES channel_instances(id) ON DELETE CASCADE, + agent_id TEXT NOT NULL REFERENCES agents(id) ON DELETE CASCADE, + user_id VARCHAR(255) NOT NULL DEFAULT '', + item_hash VARCHAR(128) NOT NULL, + item_type VARCHAR(64) NOT NULL, + summary TEXT NOT NULL, + topics TEXT NOT NULL DEFAULT '[]', + entities TEXT NOT NULL DEFAULT '[]', + confidence REAL NOT NULL DEFAULT 0, + source_id VARCHAR(255) NOT NULL DEFAULT '', + status VARCHAR(32) NOT NULL DEFAULT 'pending_review', + approved_by VARCHAR(255) NOT NULL DEFAULT '', + approved_at TEXT, + rejected_by VARCHAR(255) NOT NULL DEFAULT '', + rejected_at TEXT, + deleted_at TEXT, + written_at TEXT, + episodic_id VARCHAR(64) NOT NULL DEFAULT '', + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), + updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), + UNIQUE (tenant_id, run_id, item_hash) +); + +CREATE INDEX IF NOT EXISTS idx_channel_memory_items_channel_status + ON channel_memory_extraction_items(tenant_id, channel_instance_id, status, created_at DESC); +CREATE INDEX IF NOT EXISTS idx_channel_memory_items_run + ON channel_memory_extraction_items(tenant_id, run_id); + -- ============================================================ -- Table: channel_contacts -- ============================================================ diff --git a/internal/store/stores.go b/internal/store/stores.go index e48ef9b8..2b3b07b6 100644 --- a/internal/store/stores.go +++ b/internal/store/stores.go @@ -21,6 +21,7 @@ type Stores struct { Teams TeamStore BuiltinTools BuiltinToolStore PendingMessages PendingMessageStore + ChannelMemory ChannelMemoryExtractionStore KnowledgeGraph KnowledgeGraphStore Contacts ContactStore Activity ActivityStore diff --git a/internal/upgrade/version.go b/internal/upgrade/version.go index 7e689893..af197a28 100644 --- a/internal/upgrade/version.go +++ b/internal/upgrade/version.go @@ -2,4 +2,4 @@ package upgrade // RequiredSchemaVersion is the schema migration version this binary requires. // Bump this whenever adding a new SQL migration file. -const RequiredSchemaVersion uint = 75 +const RequiredSchemaVersion uint = 76 diff --git a/migrations/000076_channel_memory_extraction.down.sql b/migrations/000076_channel_memory_extraction.down.sql new file mode 100644 index 00000000..2626649b --- /dev/null +++ b/migrations/000076_channel_memory_extraction.down.sql @@ -0,0 +1,2 @@ +DROP TABLE IF EXISTS channel_memory_extraction_items; +DROP TABLE IF EXISTS channel_memory_extraction_runs; diff --git a/migrations/000076_channel_memory_extraction.up.sql b/migrations/000076_channel_memory_extraction.up.sql new file mode 100644 index 00000000..e454b8a3 --- /dev/null +++ b/migrations/000076_channel_memory_extraction.up.sql @@ -0,0 +1,62 @@ +CREATE TABLE IF NOT EXISTS channel_memory_extraction_runs ( + id UUID PRIMARY KEY, + tenant_id UUID NOT NULL REFERENCES tenants(id) ON DELETE CASCADE, + channel_instance_id UUID NOT NULL REFERENCES channel_instances(id) ON DELETE CASCADE, + channel_name VARCHAR(255) NOT NULL, + agent_id UUID NOT NULL REFERENCES agents(id) ON DELETE CASCADE, + user_id VARCHAR(255) NOT NULL DEFAULT '', + history_key VARCHAR(255) NOT NULL, + trigger VARCHAR(32) NOT NULL DEFAULT 'scheduled', + status VARCHAR(32) NOT NULL DEFAULT 'pending', + source_start_id VARCHAR(255) NOT NULL DEFAULT '', + source_end_id VARCHAR(255) NOT NULL DEFAULT '', + source_start_at TIMESTAMPTZ, + source_end_at TIMESTAMPTZ, + message_count INTEGER NOT NULL DEFAULT 0, + redaction_count INTEGER NOT NULL DEFAULT 0, + redaction_types JSONB NOT NULL DEFAULT '[]'::jsonb, + item_count INTEGER NOT NULL DEFAULT 0, + error_message TEXT NOT NULL DEFAULT '', + started_at TIMESTAMPTZ, + completed_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + UNIQUE (tenant_id, channel_instance_id, history_key, source_start_id, source_end_id) +); + +CREATE INDEX IF NOT EXISTS idx_channel_memory_runs_channel + ON channel_memory_extraction_runs(tenant_id, channel_instance_id, created_at DESC); +CREATE INDEX IF NOT EXISTS idx_channel_memory_runs_status + ON channel_memory_extraction_runs(tenant_id, status, created_at DESC); + +CREATE TABLE IF NOT EXISTS channel_memory_extraction_items ( + id UUID PRIMARY KEY, + tenant_id UUID NOT NULL REFERENCES tenants(id) ON DELETE CASCADE, + run_id UUID NOT NULL REFERENCES channel_memory_extraction_runs(id) ON DELETE CASCADE, + channel_instance_id UUID NOT NULL REFERENCES channel_instances(id) ON DELETE CASCADE, + agent_id UUID NOT NULL REFERENCES agents(id) ON DELETE CASCADE, + user_id VARCHAR(255) NOT NULL DEFAULT '', + item_hash VARCHAR(128) NOT NULL, + item_type VARCHAR(64) NOT NULL, + summary TEXT NOT NULL, + topics JSONB NOT NULL DEFAULT '[]'::jsonb, + entities JSONB NOT NULL DEFAULT '[]'::jsonb, + confidence DOUBLE PRECISION NOT NULL DEFAULT 0, + source_id VARCHAR(255) NOT NULL DEFAULT '', + status VARCHAR(32) NOT NULL DEFAULT 'pending_review', + approved_by VARCHAR(255) NOT NULL DEFAULT '', + approved_at TIMESTAMPTZ, + rejected_by VARCHAR(255) NOT NULL DEFAULT '', + rejected_at TIMESTAMPTZ, + deleted_at TIMESTAMPTZ, + written_at TIMESTAMPTZ, + episodic_id VARCHAR(64) NOT NULL DEFAULT '', + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + UNIQUE (tenant_id, run_id, item_hash) +); + +CREATE INDEX IF NOT EXISTS idx_channel_memory_items_channel_status + ON channel_memory_extraction_items(tenant_id, channel_instance_id, status, created_at DESC); +CREATE INDEX IF NOT EXISTS idx_channel_memory_items_run + ON channel_memory_extraction_items(tenant_id, run_id); diff --git a/plans/260528-1844-passive-channel-memory-extraction/phase-01-research-and-characterization-tests.md b/plans/260528-1844-passive-channel-memory-extraction/phase-01-research-and-characterization-tests.md new file mode 100644 index 00000000..51cbfe56 --- /dev/null +++ b/plans/260528-1844-passive-channel-memory-extraction/phase-01-research-and-characterization-tests.md @@ -0,0 +1,87 @@ +--- +phase: 1 +title: "Research and Characterization Tests" +status: complete +priority: P1 +effort: "1d" +dependencies: [] +--- + +# Phase 1: Research and Characterization Tests + +## Context Links + +- Issue: `digitopvn/goclaw#64` +- Existing docs: `docs/06-store-data-model.md`, `docs/07-bootstrap-skills-memory.md`, `docs/09-security.md`, `docs/18-http-api.md` +- Channel buffer: `internal/channels/history.go`, `internal/store/pending_message_store.go` +- Memory/KG: `internal/consolidation/episodic_worker.go`, `internal/consolidation/semantic_worker.go` + +## Overview + +Lock current behavior with tests before changing schemas or workers. This phase proves the feature can reuse existing channel pending-message buffers and Memory/KG pipeline without breaking session memory. + +## Requirements + +- Functional: document exact current data flow for group pending messages, channel instance config update, episodic creation, KG extraction, and activity logging. +- Non-functional: tests first, no implementation until expected failures describe the desired behavior. + +## Architecture + +No runtime architecture added yet. The output is a verified map and failing tests for v1 contracts: + +1. channel instance config can carry `passive_memory` +2. pending messages are tenant/channel/history-key scoped +3. extraction candidates exclude DMs/private history in v1 +4. session-created episodic flow remains unchanged + +## Related Code Files + +- Read: `internal/channels/history.go` +- Read: `internal/channels/history_compaction.go` +- Read: `internal/store/pending_message_store.go` +- Read: `internal/store/channel_instance_store.go` +- Read: `internal/consolidation/episodic_worker.go` +- Read: `internal/consolidation/semantic_worker.go` +- Read: `internal/http/channel_instances.go` +- Create: `internal/consolidation/channel_memory_config_test.go` +- Create: `internal/consolidation/channel_memory_worker_test.go` +- Create: `internal/http/channel_memory_handlers_test.go` +- Create or extend: store tests under `internal/store/pg/` and `internal/store/sqlitestore/` + +## Implementation Steps + +1. Re-scout channel adapters to list which currently call `PendingHistory.Record`. +2. Re-scout `channel_pending_messages` PG/SQLite implementations and confirm tenant predicates. +3. Write failing config parser tests for default-disabled `passive_memory`. +4. Write failing candidate-selection tests: group-only, enabled-only, interval-or-cap trigger, minimum useful content gate. +5. Write failing redaction tests using representative secret/PII strings. +6. Write failing API tests for unauthorized settings/run/review actions. +7. Write a short `plans/.../reports/scout-report.md` with verified file/line anchors. + +## Todo List + +- [ ] Enumerate all pending-history recording call sites. +- [ ] Enumerate HTTP + WS channel instance update surfaces. +- [ ] Add failing config parser tests. +- [ ] Add failing worker trigger tests. +- [ ] Add failing redaction tests. +- [ ] Add failing admin API permission tests. +- [ ] Save scout report under this plan. + +## Success Criteria + +- [ ] Tests fail for missing passive-memory config, worker, queue, and handlers. +- [ ] Scout report lists exact files to modify and no fabricated APIs. +- [ ] Plan assumptions match live code after re-grep. + +## Risk Assessment + +Main risk is planning from stale scout notes. Mitigation: every claim in later phases must cite paths found in this phase, especially schema versions and route wiring. + +## Security Considerations + +Tests must include cross-tenant pending-message isolation and reject non-admin write access before any handler implementation exists. + +## Next Steps + +Proceed to schema/store only after tests describe the contracts. diff --git a/plans/260528-1844-passive-channel-memory-extraction/phase-02-schema-and-store-contracts.md b/plans/260528-1844-passive-channel-memory-extraction/phase-02-schema-and-store-contracts.md new file mode 100644 index 00000000..91d3fe0c --- /dev/null +++ b/plans/260528-1844-passive-channel-memory-extraction/phase-02-schema-and-store-contracts.md @@ -0,0 +1,100 @@ +--- +phase: 2 +title: "Schema and Store Contracts" +status: complete +priority: P1 +effort: "2d" +dependencies: [1] +--- + +# Phase 2: Schema and Store Contracts + +## Context Links + +- Phase 1 tests and scout report +- PG migration latest during planning: `000073_secure_cli_credential_type.up.sql` +- PG schema version: `internal/upgrade/version.go` +- SQLite schema: `internal/store/sqlitestore/schema.sql`, `internal/store/sqlitestore/schema.go` +- Store registry: `internal/store/stores.go`, `internal/store/pg/factory.go`, `internal/store/sqlitestore/factory.go` + +## Overview + +Add durable run and review-item stores for passive extraction. Keep channel settings inside `channel_instances.config` to avoid another settings subsystem. + +## Requirements + +- Functional: persist extraction runs, pending extracted items, approval/rejection/deletion state, source ranges, redaction stats, and write metadata. +- Non-functional: tenant-scoped SQL, PG + SQLite parity, deterministic source IDs, no raw message body in extraction item after redaction. + +## Architecture + +Use two new store concepts: + +1. `ChannelMemoryExtractionRun`: one batch attempt per channel/history key. +2. `ChannelMemoryExtractionItem`: one extracted candidate fact or KG relation bundle pending review. + +`channel_instances.config.passive_memory` carries settings. Store layer exposes typed config helpers so HTTP/WS/UI do not mutate arbitrary JSON blindly. + +## Related Code Files + +- Create: `migrations/000074_channel_memory_extraction.up.sql` +- Create: `migrations/000074_channel_memory_extraction.down.sql` +- Modify: `internal/upgrade/version.go` +- Modify: `internal/store/sqlitestore/schema.sql` +- Modify: `internal/store/sqlitestore/schema.go` +- Create: `internal/store/channel_memory_extraction_store.go` +- Create: `internal/store/pg/channel_memory_extraction.go` +- Create: `internal/store/sqlitestore/channel_memory_extraction.go` +- Modify: `internal/store/stores.go` +- Modify: `internal/store/pg/factory.go` +- Modify: `internal/store/sqlitestore/factory.go` + +## Implementation Steps + +1. Add PG migration `000074` with tenant-scoped tables and indexes: + - `channel_memory_extraction_runs` + - `channel_memory_extraction_items` +2. Add unique constraints for idempotency: + - run: `(tenant_id, channel_instance_id, history_key, source_start_id, source_end_id)` + - item: `(tenant_id, run_id, item_hash)` +3. Add statuses: + - run: `pending`, `running`, `completed`, `failed`, `cancelled` + - item: `pending_review`, `approved`, `rejected`, `written`, `deleted` +4. Add source metadata fields: channel instance ID/name, history key, platform message ID range, time range, extractor run ID. +5. Add redaction summary fields: count, categories, skipped reason. +6. Add SQLite schema and migration map patch; bump `SchemaVersion` from 42 to 43. +7. Bump PG `RequiredSchemaVersion` from 73 to 74. +8. Implement PG and SQLite stores using existing scope patterns. +9. Wire stores into `store.Stores` and factories. +10. Make Phase 1 store tests pass. + +## Todo List + +- [ ] Add PG migration + version bump. +- [ ] Add SQLite schema + migration map + version bump. +- [ ] Add store interface types. +- [ ] Add PG store implementation. +- [ ] Add SQLite store implementation. +- [ ] Wire factories. +- [ ] Add idempotency tests. +- [ ] Add tenant isolation tests. + +## Success Criteria + +- [ ] Fresh PG migration applies and rolls back. +- [ ] Fresh SQLite DB contains tables. +- [ ] Existing SQLite DB migrates 42 to 43. +- [ ] Store tests pass for create/list/update/claim/status transitions. +- [ ] No SQL string concatenation with user input. + +## Risk Assessment + +Migration conflict risk is high on active `dev`. Before implementation, re-run `ls migrations/*.up.sql | sort | tail` and renumber if another migration claims `000074`. + +## Security Considerations + +All read/write methods require tenant context unless explicitly cross-tenant. Review item payload must store redacted text only; raw body stays in existing pending-message table subject to retention. + +## Next Steps + +Worker can claim runs only after store contracts exist. diff --git a/plans/260528-1844-passive-channel-memory-extraction/phase-03-extractor-worker-and-redaction.md b/plans/260528-1844-passive-channel-memory-extraction/phase-03-extractor-worker-and-redaction.md new file mode 100644 index 00000000..5474d9d5 --- /dev/null +++ b/plans/260528-1844-passive-channel-memory-extraction/phase-03-extractor-worker-and-redaction.md @@ -0,0 +1,114 @@ +--- +phase: 3 +title: "Extractor Worker and Redaction" +status: complete +priority: P1 +effort: "2d" +dependencies: [1, 2] +--- + +# Phase 3: Extractor Worker and Redaction + +## Context Links + +- Existing compaction: `internal/channels/history_compaction.go` +- Existing KG extractor: `internal/knowledgegraph/extractor.go` +- Background provider resolver: `internal/providerresolve` +- Usage caps: `internal/usage/caps` +- Runtime wiring: `cmd/gateway.go` + +## Overview + +Build the passive extraction worker. It scans enabled channel configs, selects eligible message batches, redacts sensitive content, calls LLM extraction/summarization once per batch, and stores review items. + +## Requirements + +- Functional: interval/message-cap/manual trigger, group-only v1, minimum useful content gate, redaction before durable writes, retry-safe run status. +- Non-functional: bounded worker concurrency, cost controlled by batching, no panic on provider failure, graceful shutdown. + +## Architecture + +Add `internal/consolidation/channel_memory_worker.go` or a focused `internal/channelmemory/` package if the file would exceed 200 LOC. + +Data flow: + +1. scan `ChannelInstanceStore` for enabled channels with `config.passive_memory.enabled=true` +2. inspect `PendingMessageStore.ListGroups` +3. choose groups where interval elapsed or message cap reached +4. load `PendingMessageStore.ListByKey` +5. redact +6. create/claim extraction run +7. call extractor prompt for durable facts only +8. persist `pending_review` items +9. log activity + `slog.Info/Warn` + +## Related Code Files + +- Create: `internal/consolidation/channel_memory_config.go` +- Create: `internal/consolidation/channel_memory_redactor.go` +- Create: `internal/consolidation/channel_memory_worker.go` +- Create: `internal/consolidation/channel_memory_prompt.go` +- Modify: `internal/consolidation/workers.go` or gateway wiring if worker stays separate from existing worker bundle +- Modify: `cmd/gateway.go` +- Modify: `cmd/gateway_managed.go` if managed gateway has separate wiring + +## Implementation Steps + +1. Implement `PassiveMemoryConfig` parser with defaults: + - disabled + - review mode true + - interval 360 minutes + - cap 100 messages + - retention 168 hours + - allowed types default set +2. Implement redactor with tests for: + - API keys/tokens/passwords + - env-like secrets + - connection strings + - payment/banking numbers + - phone/address-like PII + - configured exclude user IDs and regex patterns +3. Implement candidate selector with tests: + - skip disabled + - skip DM/private keys by v1 policy + - skip too few useful messages + - trigger on interval or message cap +4. Implement LLM prompt and parser for durable facts: + - output JSON array of items + - allowed types only + - confidence score + - concise memory text + - optional entity/relation hints +5. Use `usageCaps.Chat` with purpose `channel-memory-extraction`. +6. Store run/item status transitions. +7. Emit audit activity for run started/completed/failed. +8. Wire worker start/stop in gateway with conservative ticker and concurrency. + +## Todo List + +- [ ] Config parser tests pass. +- [ ] Redactor tests pass. +- [ ] Candidate selector tests pass. +- [ ] Worker run lifecycle tests pass. +- [ ] Provider failure leaves run `failed` and does not block gateway. +- [ ] Usage cap purpose covered. +- [ ] Gateway wiring starts worker only when channels + stores are available. + +## Success Criteria + +- [ ] Worker creates review items from eligible group messages. +- [ ] No raw unredacted message text stored in new tables. +- [ ] Duplicate scans do not create duplicate runs/items. +- [ ] Provider failures are non-fatal and auditable. + +## Risk Assessment + +The redactor can false-positive and reduce extraction quality. Accept for v1. Better to drop sensitive content than store it. + +## Security Considerations + +Prompt must explicitly reject secrets and low-confidence guesses. Logs must include counts and IDs, not raw message body. + +## Next Steps + +Approved review items can write to Memory/KG after queue semantics are stable. diff --git a/plans/260528-1844-passive-channel-memory-extraction/phase-04-memory-kg-review-queue-integration.md b/plans/260528-1844-passive-channel-memory-extraction/phase-04-memory-kg-review-queue-integration.md new file mode 100644 index 00000000..bea53fe8 --- /dev/null +++ b/plans/260528-1844-passive-channel-memory-extraction/phase-04-memory-kg-review-queue-integration.md @@ -0,0 +1,99 @@ +--- +phase: 4 +title: "Memory KG Review Queue Integration" +status: complete +priority: P1 +effort: "2d" +dependencies: [2, 3] +--- + +# Phase 4: Memory KG Review Queue Integration + +## Context Links + +- Episodic store: `internal/store/episodic_store.go`, `internal/store/pg/episodic_summaries.go`, `internal/store/sqlitestore/episodic.go` +- Semantic worker: `internal/consolidation/semantic_worker.go` +- KG store: `internal/store/knowledge_graph_store.go` +- Activity store: `internal/store/activity_store.go` + +## Overview + +Add approve/reject/delete semantics and connect approved channel extraction items to existing episodic Memory and KG ingestion. Keep review queue default-on. + +## Requirements + +- Functional: list pending items, approve, reject, delete, manual write, idempotent re-approve, source metadata preserved. +- Non-functional: approved summaries become searchable like episodic memory; KG receives only approved/redacted text; all writes are tenant/agent scoped. + +## Architecture + +Approval writes one episodic summary with: + +- `SourceType`: `channel` +- `SourceID`: deterministic `channel::` +- `SessionKey`: synthetic, e.g. `channel::` +- `Summary`: approved redacted memory text +- `KeyTopics`: extracted normalized topics/entities +- `UserID`: approving admin or configured channel agent owner user, decided in implementation after re-checking current auth model + +Then publish `EventEpisodicCreated` so `semanticWorker` extracts KG facts through the existing path. + +## Related Code Files + +- Modify: `internal/consolidation/episodic_worker.go` only if helpers can be reused without changing session behavior. +- Create: `internal/consolidation/channel_memory_writer.go` +- Modify: `internal/eventbus/event_types.go` only if channel-specific payload is genuinely needed; prefer existing `EventEpisodicCreated`. +- Modify: new `ChannelMemoryExtractionStore` from Phase 2. +- Create: tests in `internal/consolidation/channel_memory_writer_test.go`. + +## Implementation Steps + +1. Write failing tests for approve idempotency: + - first approve creates episodic and marks item written/approved + - second approve no-ops + - rejected/deleted item cannot be approved unless restored is explicitly implemented, which v1 does not need +2. Implement writer service that depends on: + - `ChannelMemoryExtractionStore` + - `EpisodicStore` + - `DomainEventBus` + - `ActivityStore` +3. Create episodic summary from approved item with deterministic source ID. +4. Publish existing `EventEpisodicCreated` with redacted summary. +5. Record activity: + - `channel_memory.item.approved` + - `channel_memory.item.rejected` + - `channel_memory.item.deleted` + - `channel_memory.item.written` +6. Add delete/forget behavior: + - v1 deletes queued item if not written + - if written, mark deleted and delete associated episodic summary when source ID matches + - KG deletion for already-ingested facts may be best-effort by source metadata if existing KG store supports it; otherwise flag as v1 limitation in UI/docs + +## Todo List + +- [ ] Add writer service tests. +- [ ] Implement approve/reject/delete transitions. +- [ ] Implement episodic write. +- [ ] Publish existing semantic event. +- [ ] Add activity logs. +- [ ] Verify session episodic tests remain unchanged. + +## Success Criteria + +- [ ] Approved item appears in episodic search. +- [ ] KG extraction path is triggered through existing event bus. +- [ ] Re-approve cannot duplicate episodic/KG data. +- [ ] Reject/delete are auditable. +- [ ] Tenant mismatch tests fail closed. + +## Risk Assessment + +Deleting KG facts after approval may require source-level relation tracking not present today. If exact deletion is not supported, v1 must be honest: delete queued item + episodic source, and mark KG cleanup as best-effort/manual dedup. + +## Security Considerations + +Approval endpoint must require tenant admin for tenant-scoped channel rows. Master-scope owner is not required because rows are tenant-scoped. + +## Next Steps + +Expose the review queue and settings through API/UI. diff --git a/plans/260528-1844-passive-channel-memory-extraction/phase-05-channel-settings-ui-and-admin-controls.md b/plans/260528-1844-passive-channel-memory-extraction/phase-05-channel-settings-ui-and-admin-controls.md new file mode 100644 index 00000000..a793234b --- /dev/null +++ b/plans/260528-1844-passive-channel-memory-extraction/phase-05-channel-settings-ui-and-admin-controls.md @@ -0,0 +1,111 @@ +--- +phase: 5 +title: "Channel Settings UI and Admin Controls" +status: complete +priority: P2 +effort: "2d" +dependencies: [2, 3, 4] +--- + +# Phase 5: Channel Settings UI and Admin Controls + +## Context Links + +- HTTP channel handler: `internal/http/channel_instances.go` +- WS channel methods: `internal/gateway/methods/channel_instances.go` +- Channel UI: `ui/web/src/pages/channels/channels-page.tsx`, `ui/web/src/pages/channels/channel-detail/` +- UI types: `ui/web/src/types/channel.ts` +- i18n: `ui/web/src/i18n/locales/{en,vi,zh}/channels.json` + +## Overview + +Expose passive memory settings and review queue in channel detail. Keep the UI operator-focused: no global dashboard in v1. + +## Requirements + +- Functional: configure, manually run extraction, show last run, show counts, approve/reject/delete pending items. +- Non-functional: mobile-safe controls, no raw secret/PII display beyond already-redacted item text, i18n in EN/VI/ZH. + +## Architecture + +HTTP endpoints under channel instance scope: + +- `GET /v1/channels/instances/{id}/memory-extraction` +- `PUT /v1/channels/instances/{id}/memory-extraction/settings` +- `POST /v1/channels/instances/{id}/memory-extraction/run` +- `GET /v1/channels/instances/{id}/memory-extraction/items` +- `POST /v1/channels/instances/{id}/memory-extraction/items/{itemID}/approve` +- `POST /v1/channels/instances/{id}/memory-extraction/items/{itemID}/reject` +- `DELETE /v1/channels/instances/{id}/memory-extraction/items/{itemID}` + +WS methods are optional in v1. Prefer HTTP because current web hooks already use HTTP for channel instances. + +## Related Code Files + +- Modify: `internal/http/channel_instances.go` +- Create: `internal/http/channel_memory_extraction_handlers.go` +- Create: `internal/http/channel_memory_extraction_handlers_test.go` +- Modify: `ui/web/src/types/channel.ts` +- Create: `ui/web/src/pages/channels/channel-detail/passive-memory-section.tsx` +- Create: `ui/web/src/pages/channels/hooks/use-channel-memory-extraction.ts` +- Modify: `ui/web/src/pages/channels/channel-detail/channel-detail-page.tsx` +- Modify: locale files in `ui/web/src/i18n/locales/en/`, `vi/`, `zh/` + +## Implementation Steps + +1. Add HTTP handler tests first: + - viewer can read status/items + - admin required for settings/run/approve/reject/delete + - tenant mismatch returns not found/forbidden without leaking row existence +2. Add settings validation: + - interval min/max + - message cap min/max + - retention min/max + - allowed types from enum + - exclude patterns bounded length/count +3. Add routes to `ChannelInstancesHandler.RegisterRoutes`. +4. Implement UI hook with TanStack Query and invalidation on mutation. +5. Build `PassiveMemorySection`: + - toggle + - interval and cap numeric inputs + - review mode toggle + - allowed type checkboxes + - last run summary + - manual run button + - pending extracted items list + - approve/reject/delete buttons +6. Add i18n keys to EN/VI/ZH channel namespace. +7. Ensure mobile rules: + - inputs use `text-base md:text-sm` + - tables/lists horizontally safe + - buttons have sufficient hit area + +## Todo List + +- [ ] Add HTTP route tests. +- [ ] Implement handler validation. +- [ ] Add UI hook. +- [ ] Add channel detail section. +- [ ] Add i18n strings. +- [ ] Add basic UI tests if existing setup supports it; otherwise rely on build + manual browser verification. + +## Success Criteria + +- [ ] Admin can enable passive memory for a channel. +- [ ] Admin can force a run. +- [ ] Pending items appear after extraction. +- [ ] Admin can approve/reject/delete items. +- [ ] Viewer cannot mutate settings/items. +- [ ] Web build passes. + +## Risk Assessment + +UI can balloon. Keep it as one compact channel-detail section. Do not add global review inbox until v1 proves useful. + +## Security Considerations + +UI must show redaction status and avoid displaying raw source messages. If source preview is needed later, it must be separate permissioned work. + +## Next Steps + +Run full verification, docs, and issue handoff. diff --git a/plans/260528-1844-passive-channel-memory-extraction/phase-06-verification-documentation-and-ship-readiness.md b/plans/260528-1844-passive-channel-memory-extraction/phase-06-verification-documentation-and-ship-readiness.md new file mode 100644 index 00000000..159b8c73 --- /dev/null +++ b/plans/260528-1844-passive-channel-memory-extraction/phase-06-verification-documentation-and-ship-readiness.md @@ -0,0 +1,99 @@ +--- +phase: 6 +title: "Verification Documentation and Ship Readiness" +status: complete +priority: P1 +effort: "1d" +dependencies: [1, 2, 3, 4, 5] +--- + +# Phase 6: Verification Documentation and Ship Readiness + +## Context Links + +- Post-implementation checklist in `CLAUDE.md` +- Docs to update: `docs/05-channels-messaging.md`, `docs/06-store-data-model.md`, `docs/07-bootstrap-skills-memory.md`, `docs/09-security.md`, `docs/18-http-api.md`, `docs/project-changelog.md` +- Local verification memory: `go build ./...`, `go build -tags sqliteonly ./...`, `go vet ./...` + +## Overview + +Prove the feature is safe, documented, and ready for implementation PR review. This phase owns the final consistency sweep and release notes. + +## Requirements + +- Functional: all new tests pass; user-facing docs describe opt-in, review queue, redaction, retention, deletion limits. +- Non-functional: compile in PG and SQLite builds, no known tenant isolation gaps, no stale plan references in code comments. + +## Architecture + +No new architecture. This phase verifies all layers together: + +1. schema migration +2. store contracts +3. worker batching/redaction +4. review queue writer +5. HTTP/UI controls +6. docs + +## Related Code Files + +- Modify: `docs/05-channels-messaging.md` +- Modify: `docs/06-store-data-model.md` +- Modify: `docs/07-bootstrap-skills-memory.md` +- Modify: `docs/09-security.md` +- Modify: `docs/18-http-api.md` +- Modify: `docs/project-changelog.md` +- Possibly modify: `docs/codebase-summary.md` + +## Implementation Steps + +1. Run focused Go tests: + - `go test ./internal/consolidation` + - `go test ./internal/http` + - `go test ./internal/store/pg` + - `go test -tags sqliteonly ./internal/store/sqlitestore` +2. Run compile/static checks: + - `go build ./...` + - `go build -tags sqliteonly ./...` + - `go vet ./...` +3. Run web checks: + - `cd ui/web && pnpm build` +4. If UI changed, run local dev server and browser-check channel detail at desktop/mobile widths. +5. Update docs listed above. +6. Update `docs/project-changelog.md`. +7. Whole-plan consistency sweep: + - search code and docs for stale names such as `passiveExtraction`, `channel_memory`, `memory-extraction` + - verify route names match docs and UI hooks + - verify schema version numbers match migrations +8. Prepare PR summary with test results and privacy caveats. + +## Todo List + +- [ ] Focused tests pass. +- [ ] PG build passes. +- [ ] SQLite build passes. +- [ ] Vet passes. +- [ ] Web build passes. +- [ ] Browser verification done if UI implemented. +- [ ] Docs updated. +- [ ] Changelog updated. +- [ ] Whole-plan consistency sweep clean. + +## Success Criteria + +- [ ] Feature satisfies all issue #64 acceptance criteria selected for v1. +- [ ] No failing tests ignored. +- [ ] Any unsupported v1 behavior documented explicitly. +- [ ] PR is ready for code review. + +## Risk Assessment + +Integration tests with PG/pgvector may require local service. If unavailable, document exactly which integration tests were skipped and why; do not claim full verification. + +## Security Considerations + +Final review must include a targeted scan for raw message body persistence, secrets in logs, missing tenant predicates, and unauthorized mutation paths. + +## Next Steps + +After plan approval, execute with `/ck:cook ` or equivalent implementation workflow. diff --git a/plans/260528-1844-passive-channel-memory-extraction/plan.md b/plans/260528-1844-passive-channel-memory-extraction/plan.md new file mode 100644 index 00000000..37860b99 --- /dev/null +++ b/plans/260528-1844-passive-channel-memory-extraction/plan.md @@ -0,0 +1,92 @@ +--- +title: "Passive Channel Memory Extraction" +description: "Add disabled-by-default, reviewable passive extraction from group channel buffers into episodic Memory and Knowledge Graph." +status: complete +priority: P1 +issue: 64 +branch: "codex/issue-64-passive-channel-memory-extraction" +tags: [channels, memory, knowledge-graph, privacy, tdd, issue-64] +blockedBy: [] +blocks: [] +created: "2026-05-28T11:45:20.128Z" +createdBy: "ck:plan" +source: skill +--- + +# Passive Channel Memory Extraction + +## Overview + +Implement issue #64 as a privacy-first v1. Agents may learn durable context from group/channel conversations they already receive, but only after a channel admin enables the feature. Extraction is batched, redacted, reviewable, tenant-scoped, and writes to Memory/KG only after approval or explicit high-confidence auto-approve. + +Hard boundary: no realtime per-message extraction, no default enablement, no permanent raw full-history store, and no cross-channel memory sharing. + +## Current Codebase Anchors + +| Area | Existing anchor | +|------|-----------------| +| Channel instances | `channel_instances.config`, `internal/store/channel_instance_store.go`, `internal/http/channel_instances.go`, `internal/gateway/methods/channel_instances.go` | +| Group buffer | `channel_pending_messages`, `internal/store/pending_message_store.go`, `internal/channels/history.go`, `internal/channels/history_compaction.go` | +| Memory/KG pipeline | `internal/consolidation/episodic_worker.go`, `internal/consolidation/semantic_worker.go`, `internal/knowledgegraph/extractor.go` | +| Activity audit | `internal/store/activity_store.go`, `internal/http/activity.go`, `emitAudit(...)` call sites | +| Runtime wiring | `cmd/gateway.go`, `cmd/gateway_http_handlers.go`, `cmd/gateway_http_wiring.go`, `cmd/gateway_channels_setup.go` | +| Web UI | `ui/web/src/pages/channels/`, `ui/web/src/pages/memory/`, `ui/web/src/types/channel.ts` | +| Schema versions | PG migration `000075`; `internal/upgrade/version.go` is 75. SQLite schema version is 44. | + +## Phases + +| Phase | Name | Status | +|-------|------|--------| +| 1 | [Research and Characterization Tests](./phase-01-research-and-characterization-tests.md) | Complete | +| 2 | [Schema and Store Contracts](./phase-02-schema-and-store-contracts.md) | Complete | +| 3 | [Extractor Worker and Redaction](./phase-03-extractor-worker-and-redaction.md) | Complete | +| 4 | [Memory KG Review Queue Integration](./phase-04-memory-kg-review-queue-integration.md) | Complete | +| 5 | [Channel Settings UI and Admin Controls](./phase-05-channel-settings-ui-and-admin-controls.md) | Complete | +| 6 | [Verification Documentation and Ship Readiness](./phase-06-verification-documentation-and-ship-readiness.md) | Complete | + +## Dependencies + +- GitHub issue: `digitopvn/goclaw#64` +- Existing group pending-message persistence must stay intact for channel context. +- Existing Memory/KG workers must keep session behavior unchanged. +- Standard edition only for channels. Lite has no channels, so SQLite migrations still must compile/apply, but UI should naturally hide channel surfaces through existing edition gates. + +## Recommended Defaults + +| Setting | Default | Reason | +|---------|---------|--------| +| `enabled` | `false` | Sensitive feature; explicit opt-in only | +| `review_mode` | `true` | Prevent accidental durable PII/secret storage | +| `interval_minutes` | `360` | Cheap daily-ish cadence without long wait | +| `message_cap` | `100` | Batching per issue request | +| `retention_hours` | `168` | 7-day raw buffer cap, same spirit as pending-message TTL | +| `allowed_types` | people, projects, decisions, todos, preferences, events | Durable work context only | +| `group_only` | `true` for v1 | Avoid DM privacy ambiguity | + +## Success Criteria + +- [x] Passive extraction is disabled by default and opt-in per channel instance. +- [x] Extraction runs only by interval, message cap, or manual trigger. +- [x] Raw channel messages are read from tenant-scoped pending buffers and are not stored permanently outside existing retention. +- [x] Redaction blocks common secrets, tokens, credentials, payment/banking data, phone/address patterns, and configurable excluded users/patterns before durable writes. +- [x] Extracted items go to a review queue by default. +- [x] Approved items write to `episodic_summaries` with `source_type='channel'` and deterministic `source_id`. +- [x] KG ingestion receives only approved/redacted summaries and records source metadata. +- [x] Review reject/delete removes queued items and prevents later writes. +- [x] Tenant/channel/agent boundaries covered by focused tests and store scoping. +- [x] PG and SQLite schemas both migrate and compile. +- [x] UI exposes settings, last run status, pending items, approve/reject/delete, and manual run. + +## Key Risks + +| Risk | Mitigation | +|------|------------| +| Durable sensitive-data leak | Redaction before queue, review default, no raw full-history durable copy | +| Duplicate facts | Deterministic run/item source IDs, store-level uniqueness, reuse KG dedup worker | +| Cross-tenant leak | Tenant-scoped queries and tests before implementation | +| Cost spike | Batched triggers, minimum content gate, usage caps purpose string | +| UX overbuild | Channel detail section only in v1; no separate global console | + +## Open Questions + +- None for v1. Assumptions: review queue required, group channels first, UI in channel detail. diff --git a/ui/web/src/i18n/locales/en/channels.json b/ui/web/src/i18n/locales/en/channels.json index 091011ad..40770a51 100644 --- a/ui/web/src/i18n/locales/en/channels.json +++ b/ui/web/src/i18n/locales/en/channels.json @@ -198,6 +198,46 @@ "actions": "Actions" } }, + "passiveMemory": { + "title": "Passive memory extraction", + "description": "Extract long-lived facts from passive group traffic into reviewable memory.", + "pendingCount": "{{count}} pending", + "enable": "Enable passive extraction", + "reviewMode": "Require review before writing", + "interval": "Interval (minutes)", + "messageCap": "Message cap", + "retention": "Retention (hours)", + "minMessages": "Min messages", + "types": "Memory types", + "excludeUsers": "Excluded users", + "excludePatterns": "Excluded patterns", + "runNow": "Run now", + "save": "Save", + "saving": "Saving...", + "saved": "Memory settings saved", + "saveFailed": "Failed to save memory settings", + "runQueued": "Memory extraction started", + "runFailed": "Memory extraction failed", + "itemUpdated": "Memory item updated", + "itemFailed": "Memory item action failed", + "reviewQueue": "Review queue", + "loading": "Loading memory status...", + "noRuns": "No extraction runs yet.", + "noItems": "No memory candidates yet.", + "lastRun": "Last run", + "runStats": "{{messages}} messages, {{items}} items, {{redactions}} redactions", + "approve": "Approve", + "reject": "Reject", + "delete": "Delete", + "type": { + "people": "People", + "projects": "Projects", + "decisions": "Decisions", + "todos": "Todos", + "preferences": "Preferences", + "events": "Events" + } + }, "timeline": { "title": "Timeline", "firstFailed": "First failed", diff --git a/ui/web/src/i18n/locales/vi/channels.json b/ui/web/src/i18n/locales/vi/channels.json index 40feea90..3513710c 100644 --- a/ui/web/src/i18n/locales/vi/channels.json +++ b/ui/web/src/i18n/locales/vi/channels.json @@ -197,6 +197,46 @@ "actions": "Thao tác" } }, + "passiveMemory": { + "title": "Trích xuất bộ nhớ thụ động", + "description": "Trích xuất các sự kiện dài hạn từ traffic nhóm vào hàng chờ duyệt bộ nhớ.", + "pendingCount": "{{count}} chờ duyệt", + "enable": "Bật trích xuất thụ động", + "reviewMode": "Yêu cầu duyệt trước khi ghi", + "interval": "Chu kỳ (phút)", + "messageCap": "Ngưỡng tin nhắn", + "retention": "Giữ lại (giờ)", + "minMessages": "Tin nhắn tối thiểu", + "types": "Loại bộ nhớ", + "excludeUsers": "User loại trừ", + "excludePatterns": "Pattern loại trừ", + "runNow": "Chạy ngay", + "save": "Lưu", + "saving": "Đang lưu...", + "saved": "Đã lưu cấu hình bộ nhớ", + "saveFailed": "Lưu cấu hình bộ nhớ thất bại", + "runQueued": "Đã bắt đầu trích xuất bộ nhớ", + "runFailed": "Trích xuất bộ nhớ thất bại", + "itemUpdated": "Đã cập nhật mục bộ nhớ", + "itemFailed": "Thao tác mục bộ nhớ thất bại", + "reviewQueue": "Hàng chờ duyệt", + "loading": "Đang tải trạng thái bộ nhớ...", + "noRuns": "Chưa có lần trích xuất nào.", + "noItems": "Chưa có ứng viên bộ nhớ.", + "lastRun": "Lần chạy gần nhất", + "runStats": "{{messages}} tin nhắn, {{items}} mục, {{redactions}} redaction", + "approve": "Duyệt", + "reject": "Từ chối", + "delete": "Xóa", + "type": { + "people": "Con người", + "projects": "Dự án", + "decisions": "Quyết định", + "todos": "Việc cần làm", + "preferences": "Sở thích", + "events": "Sự kiện" + } + }, "timeline": { "title": "Dòng thời gian", "firstFailed": "Lỗi đầu tiên", diff --git a/ui/web/src/i18n/locales/zh/channels.json b/ui/web/src/i18n/locales/zh/channels.json index 357cdd21..82a9e264 100644 --- a/ui/web/src/i18n/locales/zh/channels.json +++ b/ui/web/src/i18n/locales/zh/channels.json @@ -197,6 +197,46 @@ "actions": "操作" } }, + "passiveMemory": { + "title": "被动记忆提取", + "description": "从群组被动消息中提取长期事实，并进入可审核的记忆队列。", + "pendingCount": "{{count}} 待审核", + "enable": "启用被动提取", + "reviewMode": "写入前需要审核", + "interval": "间隔（分钟）", + "messageCap": "消息阈值", + "retention": "保留（小时）", + "minMessages": "最少消息数", + "types": "记忆类型", + "excludeUsers": "排除用户", + "excludePatterns": "排除规则", + "runNow": "立即运行", + "save": "保存", + "saving": "保存中...", + "saved": "记忆设置已保存", + "saveFailed": "保存记忆设置失败", + "runQueued": "记忆提取已启动", + "runFailed": "记忆提取失败", + "itemUpdated": "记忆项已更新", + "itemFailed": "记忆项操作失败", + "reviewQueue": "审核队列", + "loading": "正在加载记忆状态...", + "noRuns": "暂无提取运行。", + "noItems": "暂无记忆候选项。", + "lastRun": "最近运行", + "runStats": "{{messages}} 条消息，{{items}} 项，{{redactions}} 次脱敏", + "approve": "批准", + "reject": "拒绝", + "delete": "删除", + "type": { + "people": "人物", + "projects": "项目", + "decisions": "决策", + "todos": "待办", + "preferences": "偏好", + "events": "事件" + } + }, "timeline": { "title": "时间线", "firstFailed": "首次失败", diff --git a/ui/web/src/lib/query-keys.ts b/ui/web/src/lib/query-keys.ts index 00c0b0cb..3109fbee 100644 --- a/ui/web/src/lib/query-keys.ts +++ b/ui/web/src/lib/query-keys.ts @@ -39,6 +39,9 @@ export const queryKeys = { all: ["channels"] as const, list: (params: Record) => ["channels", params] as const, detail: (id: string) => ["channels", "detail", id] as const, + memoryExtraction: (id: string) => ["channels", "detail", id, "memory-extraction"] as const, + memoryExtractionItems: (id: string, params: Record) => + ["channels", "detail", id, "memory-extraction", "items", params] as const, contexts: (id: string) => ["channels", "detail", id, "contexts"] as const, contextMembers: (id: string, scopeType: string, scopeKey: string) => ["channels", "detail", id, "contexts", scopeType, scopeKey, "members"] as const, diff --git a/ui/web/src/pages/channels/channel-detail/channel-detail-page.tsx b/ui/web/src/pages/channels/channel-detail/channel-detail-page.tsx index 8bb36854..895e8149 100644 --- a/ui/web/src/pages/channels/channel-detail/channel-detail-page.tsx +++ b/ui/web/src/pages/channels/channel-detail/channel-detail-page.tsx @@ -12,6 +12,7 @@ import { ChannelGroupsTab } from "./channel-groups-tab"; import { ChannelManagersTab } from "./channel-managers-tab"; import { ChannelContextsTab } from "./channel-contexts-tab"; import { ChannelDiagnosticsCard } from "./channel-diagnostics-card"; +import { PassiveMemorySection } from "./passive-memory-section"; import { DetailPageSkeleton } from "@/components/shared/loading-skeleton"; import { useChannels } from "../hooks/use-channels"; import { channelsWithAuth } from "../channel-wizard-registry"; @@ -193,6 +194,8 @@ export function ChannelDetailPage({ )} + + diff --git a/ui/web/src/pages/channels/channel-detail/passive-memory-section-parts.tsx b/ui/web/src/pages/channels/channel-detail/passive-memory-section-parts.tsx new file mode 100644 index 00000000..253508e7 --- /dev/null +++ b/ui/web/src/pages/channels/channel-detail/passive-memory-section-parts.tsx @@ -0,0 +1,143 @@ +import { Check, Trash2, X } from "lucide-react"; +import { useTranslation } from "react-i18next"; +import { Badge } from "@/components/ui/badge"; +import { Button } from "@/components/ui/button"; +import { Input } from "@/components/ui/input"; +import { Switch } from "@/components/ui/switch"; +import { Textarea } from "@/components/ui/textarea"; +import { formatDate } from "@/lib/format"; +import type { ChannelMemoryExtractionItem } from "@/types/channel"; + +export function ToggleRow({ + label, + checked, + onCheckedChange, +}: { + label: string; + checked: boolean; + onCheckedChange: (checked: boolean) => void; +}) { + return ( + + ); +} + +export function NumberField({ + label, + value, + onChange, +}: { + label: string; + value: number; + onChange: (value: number) => void; +}) { + return ( + + ); +} + +export function TextareaBlock({ + label, + value, + onChange, +}: { + label: string; + value: string; + onChange: (value: string) => void; +}) { + return ( +