Merge pull request #116 from digitopvn/codex/issue-64-passive-channel-memory-extraction

Merged after resolving dev conflicts and preserving migration order: run timeline 000074, channel context 000075 / SQLite 44, passive channel memory 000076 / SQLite 45. Local validation passed: go build ./..., go build -tags sqliteonly ./..., go vet ./..., go test ./..., sqliteonly channelmemory/store tests, pnpm -C ui/web build. PR CI run 26706004276 passed.
This commit is contained in:
Duy /zuey/ authored and GitHub committed 2026-05-31 14:08:32 +07:00
commit 4a907135a6
47 files changed
+3354 -7

No files matched your search

+8 -1
View File
@@ -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 {
+23 -1
View File
@@ -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(),
}
}
+28 -3
View File
@@ -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.
+7
View File
@@ -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
+7
View File
@@ -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
+10
View File
@@ -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 `<<<EXTERNAL_UNTRUSTED_CONTENT>>>` 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:
+22
View File
@@ -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
+16
View File
@@ -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**
+126
View File
@@ -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
}
+81
View File
@@ -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)
}
}
+107
View File
@@ -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
}
+91
View File
@@ -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)
}
}
+49
View File
@@ -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)
}
}
+102
View File
@@ -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)
}
+163
View File
@@ -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
}
+95
View File
@@ -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)
}
}
+60
View File
@@ -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)
}
+56
View File
@@ -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)
}
}
}
+16
View File
@@ -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 {
+187
View File
@@ -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
}
@@ -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
}
@@ -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)
}
+1
View File
@@ -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),
@@ -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)
+1
View File
@@ -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),
+65 -1
View File
@@ -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,
+71
View File
@@ -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
-- ============================================================
+1
View File
@@ -21,6 +21,7 @@ type Stores struct {
Teams TeamStore
BuiltinTools BuiltinToolStore
PendingMessages PendingMessageStore
ChannelMemory ChannelMemoryExtractionStore
KnowledgeGraph KnowledgeGraphStore
Contacts ContactStore
Activity ActivityStore
+1 -1
View File
@@ -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
@@ -0,0 +1,2 @@
DROP TABLE IF EXISTS channel_memory_extraction_items;
DROP TABLE IF EXISTS channel_memory_extraction_runs;
@@ -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);
@@ -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.
@@ -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.
@@ -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.
@@ -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:<run_id>:<item_id>`
- `SessionKey`: synthetic, e.g. `channel:<channel_name>:<history_key>`
- `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.
@@ -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.
@@ -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 <plan-path>` or equivalent implementation workflow.
@@ -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.
+40
View File
@@ -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",
+40
View File
@@ -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",
+40
View File
@@ -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": "首次失败",
+3
View File
@@ -39,6 +39,9 @@ export const queryKeys = {
all: ["channels"] as const,
list: (params: Record<string, unknown>) => ["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<string, unknown>) =>
["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,
@@ -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({
</div>
)}
<PassiveMemorySection instanceId={instance.id} />
<Tabs value={activeTab} onValueChange={setActiveTab}>
<TabsList className="w-full justify-start overflow-x-auto overflow-y-hidden">
<TabsTrigger value="general">
@@ -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 (
<label className="flex items-center justify-between gap-3 rounded-lg border px-3 py-2 text-sm">
<span>{label}</span>
<Switch checked={checked} onCheckedChange={onCheckedChange} />
</label>
);
}
export function NumberField({
label,
value,
onChange,
}: {
label: string;
value: number;
onChange: (value: number) => void;
}) {
return (
<label className="space-y-1 text-xs font-medium text-muted-foreground">
<span>{label}</span>
<Input type="number" min={0} value={value} onChange={(e) => onChange(Number(e.target.value))} />
</label>
);
}
export function TextareaBlock({
label,
value,
onChange,
}: {
label: string;
value: string;
onChange: (value: string) => void;
}) {
return (
<label className="space-y-1 text-xs font-medium text-muted-foreground">
<span>{label}</span>
<Textarea size="sm" value={value} onChange={(e) => onChange(e.target.value)} />
</label>
);
}
export function RunSummary({
loading,
status,
t,
}: {
loading: boolean;
status?: {
status: string;
item_count: number;
message_count: number;
redaction_count: number;
completed_at?: string;
error_message?: string;
};
t: (key: string, opts?: Record<string, unknown>) => string;
}) {
if (loading) {
return <div className="rounded-lg border p-3 text-sm text-muted-foreground">{t("detail.passiveMemory.loading")}</div>;
}
if (!status) {
return <div className="rounded-lg border p-3 text-sm text-muted-foreground">{t("detail.passiveMemory.noRuns")}</div>;
}
return (
<div className="rounded-lg border p-3 text-sm">
<div className="flex items-center justify-between gap-2">
<span className="font-medium">{t("detail.passiveMemory.lastRun")}</span>
<Badge variant={status.status === "failed" ? "destructive" : "outline"}>{status.status}</Badge>
</div>
<div className="mt-2 text-muted-foreground">
{t("detail.passiveMemory.runStats", {
messages: status.message_count,
items: status.item_count,
redactions: status.redaction_count,
})}
</div>
{status.completed_at && <div className="mt-1 text-xs text-muted-foreground">{formatDate(status.completed_at)}</div>}
{status.error_message && <div className="mt-2 text-xs text-destructive">{status.error_message}</div>}
</div>
);
}
export function MemoryItemRow({
item,
pending,
onAction,
}: {
item: ChannelMemoryExtractionItem;
pending: boolean;
onAction: (action: "approve" | "reject" | "delete") => void;
}) {
const { t } = useTranslation("channels");
return (
<div className="rounded-lg border p-3">
<div className="flex flex-wrap items-center gap-2">
<Badge variant="info">{t(`detail.passiveMemory.type.${item.item_type}`)}</Badge>
<Badge variant={item.status === "pending_review" ? "warning" : item.status === "written" ? "success" : "outline"}>
{item.status}
</Badge>
<span className="text-xs text-muted-foreground">{Math.round(item.confidence * 100)}%</span>
</div>
<p className="mt-2 text-sm">{item.summary}</p>
<div className="mt-3 flex flex-wrap gap-2">
{item.status === "pending_review" && (
<>
<Button size="xs" variant="outline" disabled={pending} onClick={() => onAction("approve")}>
<Check />
{t("detail.passiveMemory.approve")}
</Button>
<Button size="xs" variant="outline" disabled={pending} onClick={() => onAction("reject")}>
<X />
{t("detail.passiveMemory.reject")}
</Button>
</>
)}
<Button size="xs" variant="ghost" disabled={pending} onClick={() => onAction("delete")}>
<Trash2 />
{t("detail.passiveMemory.delete")}
</Button>
</div>
</div>
);
}
@@ -0,0 +1,172 @@
import { useEffect, useMemo, useState } from "react";
import { Brain, Play, RefreshCw } from "lucide-react";
import { useTranslation } from "react-i18next";
import { Badge } from "@/components/ui/badge";
import { Button } from "@/components/ui/button";
import type { ChannelMemoryConfig } from "@/types/channel";
import { useChannelMemoryExtraction } from "../hooks/use-channel-memory-extraction";
import {
MemoryItemRow,
NumberField,
RunSummary,
TextareaBlock,
ToggleRow,
} from "./passive-memory-section-parts";
const memoryTypes = ["people", "projects", "decisions", "todos", "preferences", "events"];
const fallbackConfig: ChannelMemoryConfig = {
enabled: false,
review_mode: true,
interval_minutes: 360,
message_cap: 100,
retention_hours: 168,
allowed_types: memoryTypes,
exclude_users: [],
exclude_patterns: [],
min_messages: 5,
group_only: true,
};
interface PassiveMemorySectionProps {
instanceId: string;
}
export function PassiveMemorySection({ instanceId }: PassiveMemorySectionProps) {
const { t } = useTranslation("channels");
const { status, items, loading, saveSettings, runNow, itemAction } = useChannelMemoryExtraction(instanceId);
const [config, setConfig] = useState<ChannelMemoryConfig>(fallbackConfig);
const [excludeUsers, setExcludeUsers] = useState("");
const [excludePatterns, setExcludePatterns] = useState("");
useEffect(() => {
if (!status?.config) return;
setConfig(status.config);
setExcludeUsers((status.config.exclude_users ?? []).join("\n"));
setExcludePatterns((status.config.exclude_patterns ?? []).join("\n"));
}, [status?.config]);
const pendingItems = useMemo(() => {
return items.filter((item) => item.status === "pending_review");
}, [items]);
const updateConfig = (patch: Partial<ChannelMemoryConfig>) => {
setConfig((current) => ({ ...current, ...patch }));
};
const save = () => {
saveSettings.mutate({
...config,
exclude_users: splitLines(excludeUsers),
exclude_patterns: splitLines(excludePatterns),
group_only: true,
});
};
return (
<section className="rounded-lg border bg-card/60 p-4 shadow-xs">
<div className="flex flex-col gap-3 sm:flex-row sm:items-start sm:justify-between">
<div className="min-w-0">
<div className="flex items-center gap-2">
<Brain className="h-4 w-4 text-primary" />
<h2 className="text-sm font-semibold">{t("detail.passiveMemory.title")}</h2>
<Badge variant={config.enabled ? "success" : "outline"}>
{config.enabled ? t("enabled") : t("disabled")}
</Badge>
{pendingItems.length > 0 && (
<Badge variant="warning">
{t("detail.passiveMemory.pendingCount", { count: pendingItems.length })}
</Badge>
)}
</div>
<p className="mt-1 text-sm text-muted-foreground">
{t("detail.passiveMemory.description")}
</p>
</div>
<div className="flex gap-2">
<Button
variant="outline"
size="sm"
onClick={() => runNow.mutate()}
disabled={runNow.isPending || loading}
>
{runNow.isPending ? <RefreshCw className="animate-spin" /> : <Play />}
{t("detail.passiveMemory.runNow")}
</Button>
<Button size="sm" onClick={save} disabled={saveSettings.isPending}>
{saveSettings.isPending ? t("detail.passiveMemory.saving") : t("detail.passiveMemory.save")}
</Button>
</div>
</div>
<div className="mt-4 grid gap-4 lg:grid-cols-[1fr_1.2fr]">
<div className="space-y-4">
<ToggleRow
label={t("detail.passiveMemory.enable")}
checked={config.enabled}
onCheckedChange={(checked) => updateConfig({ enabled: checked })}
/>
<ToggleRow
label={t("detail.passiveMemory.reviewMode")}
checked={config.review_mode}
onCheckedChange={(checked) => updateConfig({ review_mode: checked })}
/>
<div className="grid grid-cols-2 gap-3">
<NumberField label={t("detail.passiveMemory.interval")} value={config.interval_minutes} onChange={(v) => updateConfig({ interval_minutes: v })} />
<NumberField label={t("detail.passiveMemory.messageCap")} value={config.message_cap} onChange={(v) => updateConfig({ message_cap: v })} />
<NumberField label={t("detail.passiveMemory.retention")} value={config.retention_hours} onChange={(v) => updateConfig({ retention_hours: v })} />
<NumberField label={t("detail.passiveMemory.minMessages")} value={config.min_messages} onChange={(v) => updateConfig({ min_messages: v })} />
</div>
<div>
<div className="mb-2 text-xs font-medium text-muted-foreground">
{t("detail.passiveMemory.types")}
</div>
<div className="flex flex-wrap gap-2">
{memoryTypes.map((type) => {
const active = config.allowed_types.includes(type);
return (
<button
key={type}
type="button"
className={`rounded-md border px-2.5 py-1 text-xs transition-colors ${active ? "border-primary bg-primary/10 text-primary" : "text-muted-foreground hover:bg-accent"}`}
onClick={() => updateConfig({ allowed_types: toggleType(config.allowed_types, type) })}
>
{t(`detail.passiveMemory.type.${type}`)}
</button>
);
})}
</div>
</div>
<TextareaBlock label={t("detail.passiveMemory.excludeUsers")} value={excludeUsers} onChange={setExcludeUsers} />
<TextareaBlock label={t("detail.passiveMemory.excludePatterns")} value={excludePatterns} onChange={setExcludePatterns} />
</div>
<div className="space-y-3">
<RunSummary loading={loading} status={status?.last_run} t={t} />
<div className="space-y-2">
<div className="text-xs font-medium text-muted-foreground">
{t("detail.passiveMemory.reviewQueue")}
</div>
{items.length === 0 ? (
<div className="rounded-lg border border-dashed p-4 text-sm text-muted-foreground">
{t("detail.passiveMemory.noItems")}
</div>
) : (
items.slice(0, 8).map((item) => (
<MemoryItemRow key={item.id} item={item} pending={itemAction.isPending} onAction={(action) => itemAction.mutate({ id: item.id, action })} />
))
)}
</div>
</div>
</div>
</section>
);
}
function toggleType(values: string[], type: string) {
return values.includes(type) ? values.filter((value) => value !== type) : [...values, type];
}
function splitLines(value: string) {
return value.split(/\n|,/).map((part) => part.trim()).filter(Boolean);
}
@@ -0,0 +1,105 @@
import { useCallback } from "react";
import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query";
import i18next from "i18next";
import { useHttp } from "@/hooks/use-ws";
import { userFriendlyError } from "@/lib/error-utils";
import { queryKeys } from "@/lib/query-keys";
import { toast } from "@/stores/use-toast-store";
import type {
ChannelMemoryConfig,
ChannelMemoryExtractionItem,
ChannelMemoryExtractionRun,
ChannelMemoryStatus,
} from "@/types/channel";
const defaultParams = {};
export function useChannelMemoryExtraction(instanceId: string | undefined) {
const http = useHttp();
const queryClient = useQueryClient();
const statusKey = queryKeys.channels.memoryExtraction(instanceId ?? "");
const invalidate = useCallback(async () => {
await Promise.all([
queryClient.invalidateQueries({ queryKey: statusKey }),
queryClient.invalidateQueries({
queryKey: queryKeys.channels.memoryExtractionItems(instanceId ?? "", defaultParams),
}),
queryClient.invalidateQueries({ queryKey: queryKeys.channels.detail(instanceId ?? "") }),
]);
}, [instanceId, queryClient, statusKey]);
const statusQuery = useQuery({
queryKey: statusKey,
queryFn: () => http.get<ChannelMemoryStatus>(`/v1/channels/instances/${instanceId}/memory-extraction`),
enabled: !!instanceId,
staleTime: 30_000,
});
const itemsQuery = useQuery({
queryKey: queryKeys.channels.memoryExtractionItems(instanceId ?? "", defaultParams),
queryFn: async () => {
const res = await http.get<{ items: ChannelMemoryExtractionItem[] }>(
`/v1/channels/instances/${instanceId}/memory-extraction/items`,
);
return res.items ?? [];
},
enabled: !!instanceId,
staleTime: 30_000,
});
const saveSettings = useMutation({
mutationFn: (config: ChannelMemoryConfig) => {
return http.put<{ config: ChannelMemoryConfig }>(
`/v1/channels/instances/${instanceId}/memory-extraction/settings`,
config,
);
},
onSuccess: async () => {
await invalidate();
toast.success(i18next.t("channels:detail.passiveMemory.saved"));
},
onError: (err) => {
toast.error(i18next.t("channels:detail.passiveMemory.saveFailed"), userFriendlyError(err));
},
});
const runNow = useMutation({
mutationFn: () => http.post<{ run: ChannelMemoryExtractionRun }>(
`/v1/channels/instances/${instanceId}/memory-extraction/run`,
),
onSuccess: async () => {
await invalidate();
toast.success(i18next.t("channels:detail.passiveMemory.runQueued"));
},
onError: (err) => {
toast.error(i18next.t("channels:detail.passiveMemory.runFailed"), userFriendlyError(err));
},
});
const itemAction = useMutation({
mutationFn: async ({ id, action }: { id: string; action: "approve" | "reject" | "delete" }) => {
const base = `/v1/channels/instances/${instanceId}/memory-extraction/items/${id}`;
if (action === "delete") {
return http.delete(base);
}
return http.post(`${base}/${action}`);
},
onSuccess: async () => {
await invalidate();
toast.success(i18next.t("channels:detail.passiveMemory.itemUpdated"));
},
onError: (err) => {
toast.error(i18next.t("channels:detail.passiveMemory.itemFailed"), userFriendlyError(err));
},
});
return {
status: statusQuery.data ?? null,
items: itemsQuery.data ?? statusQuery.data?.recent_items ?? [],
loading: statusQuery.isLoading || itemsQuery.isLoading,
saveSettings,
runNow,
itemAction,
};
}
+53
View File
@@ -51,6 +51,59 @@ export interface ChannelInstanceInput {
enabled?: boolean;
}
export interface ChannelMemoryConfig {
enabled: boolean;
review_mode: boolean;
interval_minutes: number;
message_cap: number;
retention_hours: number;
allowed_types: string[];
exclude_users?: string[];
exclude_patterns?: string[];
min_messages: number;
group_only: boolean;
}
export interface ChannelMemoryExtractionRun {
id: string;
channel_instance_id: string;
channel_name: string;
history_key: string;
trigger: string;
status: "pending" | "running" | "completed" | "failed";
message_count: number;
redaction_count: number;
item_count: number;
error_message?: string;
started_at?: string;
completed_at?: string;
created_at: string;
}
export interface ChannelMemoryExtractionItem {
id: string;
run_id: string;
item_type: string;
summary: string;
topics?: string[];
entities?: string[];
confidence: number;
status: "pending_review" | "approved" | "rejected" | "written" | "deleted";
approved_at?: string;
rejected_at?: string;
deleted_at?: string;
written_at?: string;
episodic_id?: string;
created_at: string;
}
export interface ChannelMemoryStatus {
config: ChannelMemoryConfig;
last_run?: ChannelMemoryExtractionRun;
pending_count: number;
recent_items: ChannelMemoryExtractionItem[];
}
export interface ChannelContextData {
scope_type: "channel" | "group" | "user" | "role";
scope_key: string;