From 4f926e7404ff99ef10dd4c0d1c2adf0105e13699 Mon Sep 17 00:00:00 2001 From: Wojtek Date: Thu, 21 May 2026 20:47:12 -0400 Subject: [PATCH] Add claw-wall channel-memory push ingest --- cmd/claw-wall/channel_memory.go | 173 +++++++++++++++ cmd/claw-wall/discord.go | 55 +++-- cmd/claw-wall/main.go | 57 +++-- cmd/claw-wall/main_test.go | 220 ++++++++++++++++++++ cmd/claw-wall/store.go | 80 ++++++- cmd/claw/compose_up.go | 96 ++++++--- cmd/claw/compose_up_test.go | 49 +++++ cmd/claw/spike_channel_backfill_test.go | 98 +++++++++ examples/channel-memory/.claw-describe.json | 4 + examples/channel-memory/README.md | 22 ++ examples/channel-memory/main.go | 39 +++- examples/channel-memory/main_test.go | 55 +++++ internal/pod/parser.go | 28 +++ internal/pod/parser_capabilities_test.go | 41 ++++ internal/pod/types.go | 5 + 15 files changed, 954 insertions(+), 68 deletions(-) create mode 100644 cmd/claw-wall/channel_memory.go diff --git a/cmd/claw-wall/channel_memory.go b/cmd/claw-wall/channel_memory.go new file mode 100644 index 00000000..f7b4c8a0 --- /dev/null +++ b/cmd/claw-wall/channel_memory.go @@ -0,0 +1,173 @@ +package main + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "strings" + "time" +) + +const ( + channelMemorySourceKind = "discord" + channelMemorySourceService = "claw-wall" + channelMemorySourceSurface = "discord" +) + +type channelMemoryClient struct { + ingestURL string + token string + client *http.Client +} + +type channelMemoryIngestRequest struct { + ChannelID string `json:"channel_id"` + Message channelMemoryIngestMessage `json:"message"` + Source channelMemoryIngestSource `json:"source,omitempty"` + Metadata map[string]string `json:"metadata,omitempty"` + Scope string `json:"scope,omitempty"` +} + +type channelMemoryIngestMessage struct { + ID string `json:"id"` + AuthorID string `json:"author_id,omitempty"` + AuthorName string `json:"author_name,omitempty"` + CreatedAt string `json:"created_at,omitempty"` + EditedAt string `json:"edited_at,omitempty"` + Deleted bool `json:"deleted,omitempty"` + Content string `json:"content,omitempty"` + ContentHash string `json:"content_hash,omitempty"` +} + +type channelMemoryIngestSource struct { + Kind string `json:"kind,omitempty"` + Service string `json:"service,omitempty"` + Surface string `json:"surface,omitempty"` + GuildID string `json:"guild_id,omitempty"` +} + +func newChannelMemoryClient(rawURL, token string, timeout time.Duration) (*channelMemoryClient, error) { + rawURL = strings.TrimSpace(rawURL) + if rawURL == "" { + return nil, nil + } + parsed, err := url.Parse(rawURL) + if err != nil { + return nil, fmt.Errorf("parse channel-memory ingest URL: %w", err) + } + if parsed.Scheme != "http" && parsed.Scheme != "https" { + return nil, fmt.Errorf("channel-memory ingest URL must use http or https") + } + if strings.TrimSpace(parsed.Host) == "" { + return nil, fmt.Errorf("channel-memory ingest URL must include a host") + } + if timeout <= 0 { + timeout = 2 * time.Second + } + return &channelMemoryClient{ + ingestURL: rawURL, + token: strings.TrimSpace(token), + client: &http.Client{Timeout: timeout}, + }, nil +} + +func (c *channelMemoryClient) enabled() bool { + return c != nil && strings.TrimSpace(c.ingestURL) != "" +} + +func (c *channelMemoryClient) ingestMessages(ctx context.Context, messages []wallMessage) (int, error) { + if !c.enabled() || len(messages) == 0 { + return 0, nil + } + pushed := 0 + for _, msg := range messages { + if strings.TrimSpace(msg.ChannelID) == "" || strings.TrimSpace(msg.ID) == "" { + continue + } + if err := c.ingestMessage(ctx, msg); err != nil { + return pushed, err + } + pushed++ + } + return pushed, nil +} + +func (c *channelMemoryClient) ingestMessage(ctx context.Context, msg wallMessage) error { + payload := channelMemoryPayloadForMessage(msg) + body, err := json.Marshal(payload) + if err != nil { + return err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.ingestURL, bytes.NewReader(body)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + if c.token != "" { + req.Header.Set("Authorization", "Bearer "+c.token) + } + + resp, err := c.client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 1024)) + return fmt.Errorf("channel-memory ingest returned %s: %s", resp.Status, strings.TrimSpace(string(respBody))) + } + io.Copy(io.Discard, resp.Body) + return nil +} + +func channelMemoryPayloadForMessage(msg wallMessage) channelMemoryIngestRequest { + scope := "channel:" + strings.TrimSpace(msg.ChannelID) + return channelMemoryIngestRequest{ + ChannelID: strings.TrimSpace(msg.ChannelID), + Message: channelMemoryIngestMessage{ + ID: strings.TrimSpace(msg.ID), + AuthorID: strings.TrimSpace(msg.AuthorID), + AuthorName: strings.TrimSpace(msg.Author), + CreatedAt: formatRFC3339(msg.Timestamp), + EditedAt: formatRFC3339(msg.EditedAt), + Deleted: msg.Deleted, + Content: msg.Content, + ContentHash: wallMessageContentHash(msg), + }, + Source: channelMemoryIngestSource{ + Kind: channelMemorySourceKind, + Service: channelMemorySourceService, + Surface: channelMemorySourceSurface, + }, + Metadata: map[string]string{ + "source_handle": stableSourceHandle(msg), + "visibility_scope": scope, + }, + Scope: scope, + } +} + +func wallMessageContentHash(msg wallMessage) string { + seed := strings.Join([]string{ + channelMemorySourceKind, + strings.TrimSpace(msg.ChannelID), + strings.TrimSpace(msg.ID), + msg.Content, + fmt.Sprintf("deleted=%t", msg.Deleted), + }, "\x00") + sum := sha256.Sum256([]byte(seed)) + return "sha256:" + hex.EncodeToString(sum[:]) +} + +func formatRFC3339(ts time.Time) string { + if ts.IsZero() { + return "" + } + return ts.UTC().Format(time.RFC3339) +} diff --git a/cmd/claw-wall/discord.go b/cmd/claw-wall/discord.go index c75d7371..588ce10b 100644 --- a/cmd/claw-wall/discord.go +++ b/cmd/claw-wall/discord.go @@ -27,6 +27,7 @@ type discordPoller struct { fetchLimit int backfillRetention time.Duration backfillMaxPages int + channelMemory *channelMemoryClient latestByPair map[string]string baseURL string cooldowns *rateLimitTracker @@ -34,16 +35,18 @@ type discordPoller struct { } type discordAPIMessage struct { - ID string `json:"id"` - Content string `json:"content"` - Timestamp string `json:"timestamp"` - Author discordAPIAuthor `json:"author"` - Member *discordAPIMember `json:"member,omitempty"` - Attachments []discordAttachment `json:"attachments,omitempty"` - Embeds []discordEmbed `json:"embeds,omitempty"` + ID string `json:"id"` + Content string `json:"content"` + Timestamp string `json:"timestamp"` + EditedTimestamp *string `json:"edited_timestamp,omitempty"` + Author discordAPIAuthor `json:"author"` + Member *discordAPIMember `json:"member,omitempty"` + Attachments []discordAttachment `json:"attachments,omitempty"` + Embeds []discordEmbed `json:"embeds,omitempty"` } type discordAPIAuthor struct { + ID string `json:"id"` Username string `json:"username"` GlobalName string `json:"global_name"` } @@ -160,7 +163,8 @@ func (p *discordPoller) pollOnce(ctx context.Context, logWriter io.Writer) { continue } if len(messages) > 0 { - p.store.mergeAt(target.ChannelID, messages, p.now()) + retained := p.store.mergeAt(target.ChannelID, messages, p.now()) + p.pushChannelMemory(ctx, target.ChannelID, retained, logWriter) p.recoverGapIfNeeded(ctx, target, latestID, messages, logWriter) } if strings.TrimSpace(newestID) != "" { @@ -251,7 +255,7 @@ func (p *discordPoller) backfillAll(ctx context.Context, logWriter io.Writer) { continue } p.store.setBackfillStatus(target.ChannelID, backfillStatusInProgress) - status, newestID, err := p.backfillChannel(ctx, target, p.backfillRetention, p.backfillMaxPages) + status, newestID, err := p.backfillChannel(ctx, target, p.backfillRetention, p.backfillMaxPages, logWriter) if err != nil { var rateLimitErr *discordRateLimitError if errors.As(err, &rateLimitErr) { @@ -269,7 +273,7 @@ func (p *discordPoller) backfillAll(ctx context.Context, logWriter io.Writer) { } } -func (p *discordPoller) backfillChannel(ctx context.Context, target tokenPair, retention time.Duration, maxPages int) (string, string, error) { +func (p *discordPoller) backfillChannel(ctx context.Context, target tokenPair, retention time.Duration, maxPages int, logWriter io.Writer) (string, string, error) { if retention <= 0 || maxPages <= 0 { return backfillStatusUnavailable, "", nil } @@ -294,7 +298,8 @@ func (p *discordPoller) backfillChannel(ctx context.Context, target tokenPair, r if compareSnowflakes(pageNewestID, newestID) > 0 { newestID = pageNewestID } - p.store.mergeAt(target.ChannelID, messages, p.now()) + retained := p.store.mergeAt(target.ChannelID, messages, p.now()) + p.pushChannelMemory(ctx, target.ChannelID, retained, logWriter) oldestID := messages[0].ID beforeID = oldestID @@ -313,7 +318,7 @@ func (p *discordPoller) recoverGapIfNeeded(ctx context.Context, target tokenPair return } oldestReturnedID := newestBatch[0].ID - status, err := p.backfillGap(ctx, target, previousLatestID, oldestReturnedID, p.backfillMaxPages) + status, err := p.backfillGap(ctx, target, previousLatestID, oldestReturnedID, p.backfillMaxPages, logWriter) if err != nil { var rateLimitErr *discordRateLimitError if errors.As(err, &rateLimitErr) { @@ -329,7 +334,7 @@ func (p *discordPoller) recoverGapIfNeeded(ctx context.Context, target tokenPair } } -func (p *discordPoller) backfillGap(ctx context.Context, target tokenPair, afterID, beforeID string, maxPages int) (string, error) { +func (p *discordPoller) backfillGap(ctx context.Context, target tokenPair, afterID, beforeID string, maxPages int, logWriter io.Writer) (string, error) { for page := 0; page < maxPages; page++ { messages, _, err := p.fetchMessagesBefore(ctx, target, beforeID) if err != nil { @@ -353,7 +358,8 @@ func (p *discordPoller) backfillGap(ctx context.Context, target tokenPair, after filtered = append(filtered, msg) } if len(filtered) > 0 { - p.store.mergeAt(target.ChannelID, filtered, p.now()) + retained := p.store.mergeAt(target.ChannelID, filtered, p.now()) + p.pushChannelMemory(ctx, target.ChannelID, retained, logWriter) } if reachedBoundary || len(messages) < p.fetchLimit { return backfillStatusComplete, nil @@ -363,6 +369,19 @@ func (p *discordPoller) backfillGap(ctx context.Context, target tokenPair, after return backfillStatusPartial, nil } +func (p *discordPoller) pushChannelMemory(ctx context.Context, channelID string, messages []wallMessage, logWriter io.Writer) { + if p.channelMemory == nil || len(messages) == 0 { + return + } + pushed, err := p.channelMemory.ingestMessages(ctx, messages) + if err == nil { + return + } + if logWriter != nil { + fmt.Fprintf(logWriter, "claw-wall: channel-memory ingest failed for channel %s after %d/%d messages: %v\n", channelID, pushed, len(messages), err) + } +} + func oldestMessageTime(messages []wallMessage) time.Time { var oldest time.Time for _, msg := range messages { @@ -386,13 +405,21 @@ func convertDiscordMessage(channelID string, msg discordAPIMessage) (wallMessage if err != nil { timestamp = time.Time{} } + var editedAt time.Time + if msg.EditedTimestamp != nil && strings.TrimSpace(*msg.EditedTimestamp) != "" { + if parsed, err := time.Parse(time.RFC3339, strings.TrimSpace(*msg.EditedTimestamp)); err == nil { + editedAt = parsed + } + } return wallMessage{ ID: messageID, ChannelID: channelID, + AuthorID: strings.TrimSpace(msg.Author.ID), Author: discordAuthorName(msg), Content: renderDiscordMessageContent(msg), Timestamp: timestamp, + EditedAt: editedAt, }, true } diff --git a/cmd/claw-wall/main.go b/cmd/claw-wall/main.go index 2ac394f0..1f0bf22f 100644 --- a/cmd/claw-wall/main.go +++ b/cmd/claw-wall/main.go @@ -17,15 +17,18 @@ import ( ) type config struct { - Addr string - TokenPairs string - BufferLimit int - PollInterval time.Duration - Retention time.Duration - BackfillMaxPages int - DiscordBaseURL string - ToolToken string - AgentChannelsPath string + Addr string + TokenPairs string + BufferLimit int + PollInterval time.Duration + Retention time.Duration + BackfillMaxPages int + DiscordBaseURL string + ToolToken string + AgentChannelsPath string + ChannelMemoryIngestURL string + ChannelMemoryToken string + ChannelMemoryTimeout time.Duration } func main() { @@ -55,6 +58,10 @@ func run(args []string) error { if err != nil { return fmt.Errorf("claw-wall: parse CLAW_WALL_TOKENS: %w", err) } + channelMemory, err := newChannelMemoryClient(cfg.ChannelMemoryIngestURL, cfg.ChannelMemoryToken, cfg.ChannelMemoryTimeout) + if err != nil { + return fmt.Errorf("claw-wall: configure channel-memory: %w", err) + } agentChannels, err := loadAgentChannels(cfg.AgentChannelsPath) if err != nil { return err @@ -64,6 +71,7 @@ func run(args []string) error { poller := newDiscordPoller(&http.Client{Timeout: 10 * time.Second}, store, targets, cfg.BufferLimit) poller.backfillRetention = cfg.Retention poller.backfillMaxPages = cfg.BackfillMaxPages + poller.channelMemory = channelMemory if strings.TrimSpace(cfg.DiscordBaseURL) != "" { poller.baseURL = strings.TrimRight(strings.TrimSpace(cfg.DiscordBaseURL), "/") } @@ -74,7 +82,7 @@ func run(args []string) error { server := &http.Server{ Addr: cfg.Addr, - Handler: newHandler(store, handlerConfig{toolToken: cfg.ToolToken, agentChannels: agentChannels}), + Handler: newHandler(store, handlerConfig{toolToken: cfg.ToolToken, agentChannels: agentChannels, channelMemory: channelMemory}), ReadHeaderTimeout: 10 * time.Second, } @@ -136,21 +144,32 @@ func loadConfig() (config, error) { return config{}, fmt.Errorf("claw-wall: CLAW_WALL_BACKFILL_MAX_PAGES must be non-negative") } + channelMemoryTimeout, err := envDuration("CLAW_WALL_CHANNEL_MEMORY_TIMEOUT", 2*time.Second) + if err != nil { + return config{}, err + } + if channelMemoryTimeout <= 0 { + return config{}, fmt.Errorf("claw-wall: CLAW_WALL_CHANNEL_MEMORY_TIMEOUT must be positive") + } + tokenPairs := strings.TrimSpace(os.Getenv("CLAW_WALL_TOKENS")) if tokenPairs == "" { return config{}, fmt.Errorf("claw-wall: CLAW_WALL_TOKENS is required") } return config{ - Addr: envOr("CLAW_WALL_ADDR", ":8080"), - TokenPairs: tokenPairs, - BufferLimit: bufferLimit, - PollInterval: time.Duration(pollSeconds) * time.Second, - Retention: retention, - BackfillMaxPages: backfillMaxPages, - DiscordBaseURL: strings.TrimSpace(os.Getenv("CLAW_WALL_DISCORD_BASE_URL")), - ToolToken: strings.TrimSpace(os.Getenv("CLAW_WALL_TOOL_TOKEN")), - AgentChannelsPath: envOr("CLAW_WALL_AGENT_CHANNELS_FILE", "/etc/claw-wall/agent-channels.json"), + Addr: envOr("CLAW_WALL_ADDR", ":8080"), + TokenPairs: tokenPairs, + BufferLimit: bufferLimit, + PollInterval: time.Duration(pollSeconds) * time.Second, + Retention: retention, + BackfillMaxPages: backfillMaxPages, + DiscordBaseURL: strings.TrimSpace(os.Getenv("CLAW_WALL_DISCORD_BASE_URL")), + ToolToken: strings.TrimSpace(os.Getenv("CLAW_WALL_TOOL_TOKEN")), + AgentChannelsPath: envOr("CLAW_WALL_AGENT_CHANNELS_FILE", "/etc/claw-wall/agent-channels.json"), + ChannelMemoryIngestURL: strings.TrimSpace(os.Getenv("CLAW_WALL_CHANNEL_MEMORY_INGEST_URL")), + ChannelMemoryToken: strings.TrimSpace(os.Getenv("CLAW_WALL_CHANNEL_MEMORY_TOKEN")), + ChannelMemoryTimeout: channelMemoryTimeout, }, nil } diff --git a/cmd/claw-wall/main_test.go b/cmd/claw-wall/main_test.go index cbaf06b8..1de3d93f 100644 --- a/cmd/claw-wall/main_test.go +++ b/cmd/claw-wall/main_test.go @@ -529,6 +529,9 @@ func TestLoadConfigDefaultsPollIntervalToThirtySeconds(t *testing.T) { t.Setenv("CLAW_WALL_RETENTION", "") t.Setenv("CLAW_WALL_BACKFILL_MAX_PAGES", "") t.Setenv("CLAW_WALL_DISCORD_BASE_URL", "") + t.Setenv("CLAW_WALL_CHANNEL_MEMORY_INGEST_URL", "") + t.Setenv("CLAW_WALL_CHANNEL_MEMORY_TOKEN", "") + t.Setenv("CLAW_WALL_CHANNEL_MEMORY_TIMEOUT", "") t.Setenv("CLAW_WALL_TOKENS", "chan-1:token-a") cfg, err := loadConfig() @@ -550,6 +553,9 @@ func TestLoadConfigDefaultsPollIntervalToThirtySeconds(t *testing.T) { if cfg.DiscordBaseURL != "" { t.Fatalf("expected empty discord base url by default, got %q", cfg.DiscordBaseURL) } + if cfg.ChannelMemoryIngestURL != "" || cfg.ChannelMemoryToken != "" || cfg.ChannelMemoryTimeout != 2*time.Second { + t.Fatalf("unexpected channel-memory defaults: %+v", cfg) + } } func TestLoadConfigReadsDiscordBaseURLOverride(t *testing.T) { @@ -559,6 +565,7 @@ func TestLoadConfigReadsDiscordBaseURLOverride(t *testing.T) { t.Setenv("CLAW_WALL_RETENTION", "") t.Setenv("CLAW_WALL_BACKFILL_MAX_PAGES", "") t.Setenv("CLAW_WALL_DISCORD_BASE_URL", "http://fake-discord:9000/api/v10") + t.Setenv("CLAW_WALL_CHANNEL_MEMORY_TIMEOUT", "") t.Setenv("CLAW_WALL_TOKENS", "chan-1:token-a") cfg, err := loadConfig() @@ -570,12 +577,34 @@ func TestLoadConfigReadsDiscordBaseURLOverride(t *testing.T) { } } +func TestLoadConfigReadsChannelMemoryConfig(t *testing.T) { + t.Setenv("CLAW_WALL_ADDR", "") + t.Setenv("CLAW_WALL_LIMIT", "") + t.Setenv("CLAW_WALL_POLL_INTERVAL", "") + t.Setenv("CLAW_WALL_RETENTION", "") + t.Setenv("CLAW_WALL_BACKFILL_MAX_PAGES", "") + t.Setenv("CLAW_WALL_DISCORD_BASE_URL", "") + t.Setenv("CLAW_WALL_CHANNEL_MEMORY_INGEST_URL", "http://channel-memory:8080/ingest") + t.Setenv("CLAW_WALL_CHANNEL_MEMORY_TOKEN", "memory-token") + t.Setenv("CLAW_WALL_CHANNEL_MEMORY_TIMEOUT", "750ms") + t.Setenv("CLAW_WALL_TOKENS", "chan-1:token-a") + + cfg, err := loadConfig() + if err != nil { + t.Fatalf("loadConfig: %v", err) + } + if cfg.ChannelMemoryIngestURL != "http://channel-memory:8080/ingest" || cfg.ChannelMemoryToken != "memory-token" || cfg.ChannelMemoryTimeout != 750*time.Millisecond { + t.Fatalf("unexpected channel-memory config: %+v", cfg) + } +} + func TestLoadConfigUsesPollIntervalOverride(t *testing.T) { t.Setenv("CLAW_WALL_ADDR", "") t.Setenv("CLAW_WALL_LIMIT", "") t.Setenv("CLAW_WALL_POLL_INTERVAL", "42") t.Setenv("CLAW_WALL_RETENTION", "6h") t.Setenv("CLAW_WALL_BACKFILL_MAX_PAGES", "7") + t.Setenv("CLAW_WALL_CHANNEL_MEMORY_TIMEOUT", "") t.Setenv("CLAW_WALL_TOKENS", "chan-1:token-a") cfg, err := loadConfig() @@ -728,6 +757,160 @@ func TestDiscordPollerBackfillRespectsRateLimit(t *testing.T) { } } +func TestDiscordPollerPushesBackfillToChannelMemory(t *testing.T) { + now := time.Date(2026, 5, 21, 16, 0, 0, 0, time.UTC) + messages := makeDiscordMessages(now.Add(-3*time.Minute), 3, time.Minute) + messages[0].Author.ID = "user-1000" + memory := newRecordingChannelMemoryServer(t, http.StatusAccepted) + defer memory.Close() + client, err := newChannelMemoryClient(memory.URL+"/ingest", "", time.Second) + if err != nil { + t.Fatalf("newChannelMemoryClient: %v", err) + } + discord := newDiscordMessagesServer(t, messages, nil) + defer discord.Close() + + store := newConversationStore(50, 24*time.Hour) + poller := newDiscordPoller(discord.Client(), store, []tokenPair{{ChannelID: "chan-1", Token: "token-a"}}, 50) + poller.baseURL = discord.URL + poller.now = func() time.Time { return now } + poller.backfillRetention = 24 * time.Hour + poller.backfillMaxPages = 1 + poller.channelMemory = client + + poller.backfillAll(context.Background(), io.Discard) + + got := memory.requests() + if len(got) != 3 { + t.Fatalf("expected 3 channel-memory ingests, got %d: %+v", len(got), got) + } + if got[0].ChannelID != "chan-1" || got[0].Message.ID != "1000" || got[0].Message.AuthorID != "user-1000" { + t.Fatalf("unexpected first ingest payload: %+v", got[0]) + } + if got[0].Message.ContentHash == "" || !strings.HasPrefix(got[0].Message.ContentHash, "sha256:") { + t.Fatalf("expected content hash in ingest payload: %+v", got[0].Message) + } + if got[0].Scope != "channel:chan-1" || got[0].Metadata["visibility_scope"] != "channel:chan-1" { + t.Fatalf("expected channel visibility scope, got %+v", got[0]) + } +} + +func TestDiscordPollerPushesForwardPollToChannelMemory(t *testing.T) { + now := time.Date(2026, 5, 21, 16, 0, 0, 0, time.UTC) + messages := makeDiscordMessages(now.Add(-3*time.Minute), 3, time.Minute) + memory := newRecordingChannelMemoryServer(t, http.StatusAccepted) + defer memory.Close() + client, err := newChannelMemoryClient(memory.URL+"/ingest", "", time.Second) + if err != nil { + t.Fatalf("newChannelMemoryClient: %v", err) + } + discord := newDiscordMessagesServer(t, messages, nil) + defer discord.Close() + + store := newConversationStore(50, 24*time.Hour) + poller := newDiscordPoller(discord.Client(), store, []tokenPair{{ChannelID: "chan-1", Token: "token-a"}}, 50) + poller.baseURL = discord.URL + poller.now = func() time.Time { return now } + poller.latestByPair[pairKey(tokenPair{ChannelID: "chan-1", Token: "token-a"})] = "1000" + poller.channelMemory = client + + poller.pollOnce(context.Background(), io.Discard) + + got := memory.requests() + if len(got) != 2 || got[0].Message.ID != "1001" || got[1].Message.ID != "1002" { + t.Fatalf("expected forward poll messages 1001,1002, got %+v", got) + } +} + +func TestDiscordPollerChannelMemoryFailureDoesNotBlockRawWindow(t *testing.T) { + now := time.Date(2026, 5, 21, 16, 0, 0, 0, time.UTC) + messages := makeDiscordMessages(now.Add(-time.Minute), 1, time.Minute) + memory := newRecordingChannelMemoryServer(t, http.StatusInternalServerError) + defer memory.Close() + client, err := newChannelMemoryClient(memory.URL+"/ingest", "", time.Second) + if err != nil { + t.Fatalf("newChannelMemoryClient: %v", err) + } + discord := newDiscordMessagesServer(t, messages, nil) + defer discord.Close() + + store := newConversationStore(50, 24*time.Hour) + poller := newDiscordPoller(discord.Client(), store, []tokenPair{{ChannelID: "chan-1", Token: "token-a"}}, 50) + poller.baseURL = discord.URL + poller.now = func() time.Time { return now } + poller.channelMemory = client + + var logs strings.Builder + poller.pollOnce(context.Background(), &logs) + + if !strings.Contains(logs.String(), "channel-memory ingest failed") { + t.Fatalf("expected channel-memory failure log, got %q", logs.String()) + } + got := store.tail(tailRequest{ChannelIDs: []string{"chan-1"}, Since: 24 * time.Hour, Limit: 10, Now: now}) + if len(got.Messages) != 1 || got.Messages[0].ID != "1000" { + t.Fatalf("expected raw window to retain message despite ingest failure, got %+v", got.Messages) + } +} + +func TestChannelMemoryReplayRequiresAllowedChannels(t *testing.T) { + store := newConversationStore(50) + store.merge("chan-1", []wallMessage{{ID: "100", Author: "alice", Content: "alpha", Timestamp: time.Unix(100, 0)}}) + store.merge("chan-2", []wallMessage{{ID: "200", Author: "bob", Content: "beta", Timestamp: time.Unix(200, 0)}}) + memory := newRecordingChannelMemoryServer(t, http.StatusAccepted) + defer memory.Close() + client, err := newChannelMemoryClient(memory.URL+"/ingest", "", time.Second) + if err != nil { + t.Fatalf("newChannelMemoryClient: %v", err) + } + server := httptest.NewServer(newHandler(store, handlerConfig{ + toolToken: "tool-token", + channelMemory: client, + agentChannels: map[string]map[string]struct{}{ + "trader-0": {"chan-1": {}}, + }, + })) + defer server.Close() + + req, err := http.NewRequest(http.MethodPost, server.URL+"/channel-memory/replay", strings.NewReader(`{"channels":["chan-2"]}`)) + if err != nil { + t.Fatalf("request forbidden: %v", err) + } + req.Header.Set("Authorization", "Bearer tool-token") + req.Header.Set("X-Claw-ID", "trader-0") + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("POST forbidden: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusForbidden { + body, _ := io.ReadAll(resp.Body) + t.Fatalf("expected 403, got %d: %s", resp.StatusCode, string(body)) + } + if got := memory.requests(); len(got) != 0 { + t.Fatalf("forbidden replay should not push messages, got %+v", got) + } + + req, err = http.NewRequest(http.MethodPost, server.URL+"/channel-memory/replay", strings.NewReader(`{"channels":["chan-1"]}`)) + if err != nil { + t.Fatalf("request replay: %v", err) + } + req.Header.Set("Authorization", "Bearer tool-token") + req.Header.Set("X-Claw-ID", "trader-0") + resp, err = http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("POST replay: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(resp.Body) + t.Fatalf("expected 200, got %d: %s", resp.StatusCode, string(body)) + } + got := memory.requests() + if len(got) != 1 || got[0].ChannelID != "chan-1" || got[0].Message.ID != "100" { + t.Fatalf("expected allowed replay to push chan-1/100, got %+v", got) + } +} + func TestDiscordPollerRecoversFullForwardPollGap(t *testing.T) { now := time.Date(2026, 5, 12, 18, 0, 0, 0, time.UTC) messages := makeDiscordMessages(now.Add(-4*time.Hour), 350, time.Minute) @@ -964,6 +1147,42 @@ func TestDiscordPollerResumesPollingAfterCooldownExpires(t *testing.T) { } } +type recordingChannelMemoryServer struct { + *httptest.Server + mu sync.Mutex + received []channelMemoryIngestRequest +} + +func newRecordingChannelMemoryServer(t *testing.T, status int) *recordingChannelMemoryServer { + t.Helper() + rec := &recordingChannelMemoryServer{} + rec.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost || r.URL.Path != "/ingest" { + http.Error(w, "unexpected request", http.StatusNotFound) + return + } + var payload channelMemoryIngestRequest + if err := json.NewDecoder(r.Body).Decode(&payload); err != nil { + http.Error(w, "invalid json", http.StatusBadRequest) + return + } + rec.mu.Lock() + rec.received = append(rec.received, payload) + rec.mu.Unlock() + w.WriteHeader(status) + _, _ = io.WriteString(w, `{"ok":true}`) + })) + return rec +} + +func (s *recordingChannelMemoryServer) requests() []channelMemoryIngestRequest { + s.mu.Lock() + defer s.mu.Unlock() + out := make([]channelMemoryIngestRequest, len(s.received)) + copy(out, s.received) + return out +} + func makeDiscordMessages(start time.Time, count int, step time.Duration) []discordAPIMessage { messages := make([]discordAPIMessage, 0, count) for i := 0; i < count; i++ { @@ -972,6 +1191,7 @@ func makeDiscordMessages(start time.Time, count int, step time.Duration) []disco Content: fmt.Sprintf("message-%03d", i), Timestamp: start.Add(time.Duration(i) * step).Format(time.RFC3339), Author: discordAPIAuthor{ + ID: fmt.Sprintf("user-id-%03d", i), Username: fmt.Sprintf("user-%03d", i), }, }) diff --git a/cmd/claw-wall/store.go b/cmd/claw-wall/store.go index 7279d606..3d3e3fb7 100644 --- a/cmd/claw-wall/store.go +++ b/cmd/claw-wall/store.go @@ -32,9 +32,12 @@ type wallMessage struct { ID string `json:"id"` ChannelID string `json:"channel_id"` SourceHandle string `json:"source_handle,omitempty"` + AuthorID string `json:"author_id,omitempty"` Author string `json:"author"` Content string `json:"content"` Timestamp time.Time `json:"timestamp"` + EditedAt time.Time `json:"edited_at,omitempty"` + Deleted bool `json:"deleted,omitempty"` } type channelBuffer struct { @@ -78,6 +81,7 @@ type tailResult struct { type handlerConfig struct { toolToken string agentChannels map[string]map[string]struct{} + channelMemory *channelMemoryClient } type agentChannelAllowlistFile struct { @@ -118,6 +122,17 @@ type getChannelMessagesRequest struct { Limit int `json:"limit"` } +type channelMemoryReplayRequest struct { + Channels []string `json:"channels"` + Limit int `json:"limit"` +} + +type channelMemoryReplayResponse struct { + Status string `json:"status"` + Messages int `json:"messages"` + Pushed int `json:"pushed"` +} + func newConversationStore(limit int, retentions ...time.Duration) *conversationStore { var retention time.Duration if len(retentions) > 0 { @@ -149,18 +164,18 @@ func (s *conversationStore) effectiveTime(now time.Time) time.Time { return now } -func (s *conversationStore) merge(channelID string, messages []wallMessage) { - s.mergeAt(channelID, messages, s.currentTime()) +func (s *conversationStore) merge(channelID string, messages []wallMessage) []wallMessage { + return s.mergeAt(channelID, messages, s.currentTime()) } -func (s *conversationStore) mergeAt(channelID string, messages []wallMessage, now time.Time) { +func (s *conversationStore) mergeAt(channelID string, messages []wallMessage, now time.Time) []wallMessage { if len(messages) == 0 { - return + return nil } now = s.effectiveTime(now) channelID = strings.TrimSpace(channelID) if channelID == "" { - return + return nil } s.mu.Lock() @@ -172,6 +187,7 @@ func (s *conversationStore) mergeAt(channelID string, messages []wallMessage, no s.channels[channelID] = state } + retained := make([]wallMessage, 0, len(messages)) for _, msg := range messages { if strings.TrimSpace(msg.ID) == "" { continue @@ -191,10 +207,12 @@ func (s *conversationStore) mergeAt(channelID string, messages []wallMessage, no msg.SourceHandle = stableSourceHandle(msg) state.seenIDs[msg.ID] = struct{}{} state.messages = append(state.messages, msg) + retained = append(retained, msg) } sortWallMessages(state.messages) s.trimChannelLocked(channelID, state, now) + return retained } func (s *conversationStore) setBackfillStatus(channelID, status string) { @@ -604,6 +622,30 @@ func (s *conversationStore) scan(channelIDs []string, match func(wallMessage) bo } } +func (s *conversationStore) snapshot(channelIDs []string, limit int, now time.Time) []wallMessage { + channelIDs = normalizeChannelIDs(channelIDs) + s.mu.Lock() + defer s.mu.Unlock() + now = s.effectiveTime(now) + s.trimChannelsLocked(channelIDs, now) + + messages := make([]wallMessage, 0) + for _, channelID := range channelIDs { + state := s.channels[channelID] + if state == nil { + continue + } + messages = append(messages, state.messages...) + } + sortWallMessages(messages) + if limit > 0 && len(messages) > limit { + messages = messages[len(messages)-limit:] + } + out := make([]wallMessage, len(messages)) + copy(out, messages) + return out +} + func (s *conversationStore) backfillStatusForChannelsLocked(channelIDs []string) map[string]string { channelIDs = normalizeChannelIDs(channelIDs) out := make(map[string]string, len(channelIDs)) @@ -774,6 +816,34 @@ func newHandler(store *conversationStore, cfgs ...handlerConfig) http.Handler { } writeJSON(w, http.StatusOK, store.getMessages(req)) }) + mux.HandleFunc("/channel-memory/replay", func(w http.ResponseWriter, r *http.Request) { + if !authorizeToolRequest(w, r, cfg) { + return + } + if !cfg.channelMemory.enabled() { + writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "channel_memory_disabled"}) + return + } + var req channelMemoryReplayRequest + if !decodeJSONRequest(w, r, &req) { + return + } + if len(normalizeChannelIDs(req.Channels)) == 0 { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "channels_required"}) + return + } + if disallowed := firstDisallowedToolChannel(req.Channels, r.Header.Get("X-Claw-ID"), cfg.agentChannels); disallowed != "" { + writeJSON(w, http.StatusForbidden, map[string]string{"error": "channel_not_allowed", "agent": r.Header.Get("X-Claw-ID"), "channel": disallowed}) + return + } + messages := store.snapshot(req.Channels, req.Limit, store.currentTime()) + pushed, err := cfg.channelMemory.ingestMessages(r.Context(), messages) + if err != nil { + writeJSON(w, http.StatusBadGateway, map[string]interface{}{"error": "channel_memory_ingest_failed", "pushed": pushed}) + return + } + writeJSON(w, http.StatusOK, channelMemoryReplayResponse{Status: "ok", Messages: len(messages), Pushed: pushed}) + }) return mux } diff --git a/cmd/claw/compose_up.go b/cmd/claw/compose_up.go index 81bba69f..982d6dcb 100644 --- a/cmd/claw/compose_up.go +++ b/cmd/claw/compose_up.go @@ -42,25 +42,27 @@ var composeUpDiscoverTools bool var runtimePlaceholderPattern = regexp.MustCompile(`\$\{([^}]+)\}`) const ( - conversationWallServiceName = "claw-wall" - conversationWallFeedName = "channel-context" - conversationWallAwarenessName = "channel-awareness" - conversationWallFeedTTL = 30 - conversationWallFeedSince = "24h" - conversationWallFeedLimit = 40 - conversationWallAwarenessLimit = 60 - conversationWallFeedMaxChars = 32 * 1024 - conversationWallBufferLimit = 5000 - conversationWallPollInterval = "30" - conversationWallRetention = "24h" - conversationWallBackfillPages = "25" - conversationWallInternalPort = "8080" - conversationWallDockerfile = "dockerfiles/claw-wall/Dockerfile" - conversationWallToolTokenEnv = "CLAW_WALL_TOOL_TOKEN" - conversationWallAllowlistPath = "/etc/claw-wall/agent-channels.json" - clawInternalNetworkName = "claw-internal" - historyReplayAuthService = "cllama-history" - historyReplayBaseURL = "http://cllama:8080/history" + conversationWallServiceName = "claw-wall" + conversationWallFeedName = "channel-context" + conversationWallAwarenessName = "channel-awareness" + conversationWallFeedTTL = 30 + conversationWallFeedSince = "24h" + conversationWallFeedLimit = 40 + conversationWallAwarenessLimit = 60 + conversationWallFeedMaxChars = 32 * 1024 + conversationWallBufferLimit = 5000 + conversationWallPollInterval = "30" + conversationWallRetention = "24h" + conversationWallBackfillPages = "25" + conversationWallInternalPort = "8080" + conversationWallDockerfile = "dockerfiles/claw-wall/Dockerfile" + conversationWallToolTokenEnv = "CLAW_WALL_TOOL_TOKEN" + conversationWallAllowlistPath = "/etc/claw-wall/agent-channels.json" + conversationWallMemoryIngestEnv = "CLAW_WALL_CHANNEL_MEMORY_INGEST_URL" + conversationWallMemoryTimeout = "2s" + clawInternalNetworkName = "claw-internal" + historyReplayAuthService = "cllama-history" + historyReplayBaseURL = "http://cllama:8080/history" ) var ( @@ -1816,16 +1818,21 @@ func injectConversationWall(p *pod.Pod, resolvedClaws map[string]*driver.Resolve svc.Claw.Tools = appendConversationWallToolPolicy(svc.Claw.Tools) } + wallEnv := map[string]string{ + "CLAW_WALL_TOKENS": formatConversationWallTokenPairs(tokenPairs), + "CLAW_WALL_LIMIT": strconv.Itoa(conversationWallBufferForPod(p, triggerServices)), + "CLAW_WALL_POLL_INTERVAL": envOrDefault("CLAW_WALL_POLL_INTERVAL", conversationWallPollInterval), + "CLAW_WALL_RETENTION": envOrDefault("CLAW_WALL_RETENTION", conversationWallRetention), + "CLAW_WALL_BACKFILL_MAX_PAGES": envOrDefault("CLAW_WALL_BACKFILL_MAX_PAGES", conversationWallBackfillPages), + } + if err := configureConversationWallChannelMemory(p, wallEnv); err != nil { + return err + } + p.Services[conversationWallServiceName] = &pod.Service{ - Image: resolveConversationWallImageRef(), - Environment: map[string]string{ - "CLAW_WALL_TOKENS": formatConversationWallTokenPairs(tokenPairs), - "CLAW_WALL_LIMIT": strconv.Itoa(conversationWallBufferForPod(p, triggerServices)), - "CLAW_WALL_POLL_INTERVAL": envOrDefault("CLAW_WALL_POLL_INTERVAL", conversationWallPollInterval), - "CLAW_WALL_RETENTION": envOrDefault("CLAW_WALL_RETENTION", conversationWallRetention), - "CLAW_WALL_BACKFILL_MAX_PAGES": envOrDefault("CLAW_WALL_BACKFILL_MAX_PAGES", conversationWallBackfillPages), - }, - Expose: []string{conversationWallInternalPort}, + Image: resolveConversationWallImageRef(), + Environment: wallEnv, + Expose: []string{conversationWallInternalPort}, Compose: map[string]interface{}{ "networks": []string{"claw-internal"}, "restart": "on-failure", @@ -1845,6 +1852,39 @@ func injectConversationWall(p *pod.Pod, resolvedClaws map[string]*driver.Resolve return nil } +func configureConversationWallChannelMemory(p *pod.Pod, wallEnv map[string]string) error { + if p == nil || p.ChannelMemory == nil { + return nil + } + serviceName := strings.TrimSpace(p.ChannelMemory.Service) + if serviceName == "" { + return nil + } + if _, ok := p.Services[serviceName]; !ok { + return fmt.Errorf("channel-memory service %q not found", serviceName) + } + baseURL, err := resolveServiceBaseURL(p, serviceName) + if err != nil { + return fmt.Errorf("channel-memory service %q: %w", serviceName, err) + } + wallEnv[conversationWallMemoryIngestEnv] = buildFeedURL(baseURL, "/ingest") + wallEnv["CLAW_WALL_CHANNEL_MEMORY_TIMEOUT"] = envOrDefault("CLAW_WALL_CHANNEL_MEMORY_TIMEOUT", conversationWallMemoryTimeout) + targetSvc := p.Services[serviceName] + if targetSvc.Environment == nil { + targetSvc.Environment = make(map[string]string) + } + token := strings.TrimSpace(targetSvc.Environment["CHANNEL_MEMORY_TOKEN"]) + if token == "" { + token = cllama.GenerateToken(serviceName) + targetSvc.Environment["CHANNEL_MEMORY_TOKEN"] = token + } + wallEnv["CLAW_WALL_CHANNEL_MEMORY_TOKEN"] = token + if err := ensureServiceOnNetwork(targetSvc, clawInternalNetworkName); err != nil { + return fmt.Errorf("attach channel-memory service network: %w", err) + } + return nil +} + func selectConversationWallToken(channelID, master string, candidates []conversationWallTokenCandidate) (string, error) { if len(candidates) == 0 { return "", fmt.Errorf("conversation wall injection triggered but channel %q has no eligible Discord reader token", channelID) diff --git a/cmd/claw/compose_up_test.go b/cmd/claw/compose_up_test.go index 57a105d4..d9a61c58 100644 --- a/cmd/claw/compose_up_test.go +++ b/cmd/claw/compose_up_test.go @@ -4026,6 +4026,55 @@ func TestInjectConversationWallHonorsChannelContextConfig(t *testing.T) { } } +func TestInjectConversationWallWiresChannelMemory(t *testing.T) { + t.Setenv("CLAW_WALL_RETENTION", "") + t.Setenv("CLAW_WALL_BACKFILL_MAX_PAGES", "") + t.Setenv("CLAW_WALL_CHANNEL_MEMORY_TIMEOUT", "") + + p := &pod.Pod{ + Name: "desk", + ChannelMemory: &pod.ChannelMemoryConfig{Service: "channel-memory"}, + Services: map[string]*pod.Service{ + "channel-memory": { + Image: "channel-memory:latest", + Expose: []string{"8080"}, + Compose: map[string]interface{}{ + "restart": "on-failure", + }, + }, + "trader": testConversationWallService("${TRADER_DISCORD_BOT_TOKEN}", "chan-1"), + }, + } + resolvedClaws := map[string]*driver.ResolvedClaw{ + "trader": {ServiceName: "trader", Cllama: []string{"passthrough"}}, + } + + if err := injectConversationWall(p, resolvedClaws); err != nil { + t.Fatalf("injectConversationWall: %v", err) + } + + wall := p.Services[conversationWallServiceName] + if wall == nil { + t.Fatal("expected claw-wall service") + } + if wall.Environment[conversationWallMemoryIngestEnv] != "http://channel-memory:8080/ingest" { + t.Fatalf("unexpected channel-memory ingest URL: %q", wall.Environment[conversationWallMemoryIngestEnv]) + } + if wall.Environment["CLAW_WALL_CHANNEL_MEMORY_TIMEOUT"] != conversationWallMemoryTimeout { + t.Fatalf("unexpected channel-memory timeout: %q", wall.Environment["CLAW_WALL_CHANNEL_MEMORY_TIMEOUT"]) + } + if wall.Environment["CLAW_WALL_CHANNEL_MEMORY_TOKEN"] == "" { + t.Fatal("expected claw-wall channel-memory token") + } + if p.Services["channel-memory"].Environment["CHANNEL_MEMORY_TOKEN"] != wall.Environment["CLAW_WALL_CHANNEL_MEMORY_TOKEN"] { + t.Fatalf("expected channel-memory token to match claw-wall token") + } + networks, ok := p.Services["channel-memory"].Compose["networks"].([]string) + if !ok || len(networks) != 1 || networks[0] != clawInternalNetworkName { + t.Fatalf("expected channel-memory on %s, got %#v", clawInternalNetworkName, p.Services["channel-memory"].Compose["networks"]) + } +} + func TestPrepareConversationWallRuntimeWritesAllowlistAndServiceAuth(t *testing.T) { runtimeDir := t.TempDir() p := &pod.Pod{ diff --git a/cmd/claw/spike_channel_backfill_test.go b/cmd/claw/spike_channel_backfill_test.go index 8f1bcec8..9c8cc297 100644 --- a/cmd/claw/spike_channel_backfill_test.go +++ b/cmd/claw/spike_channel_backfill_test.go @@ -87,6 +87,41 @@ func TestSpikeChannelBackfill(t *testing.T) { t.Fatalf("split host port: %v", err) } + var ( + memoryMu sync.Mutex + ingested = make(map[string]fakeChannelMemoryIngest) + ) + memoryHandler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost || r.URL.Path != "/ingest" { + http.Error(w, "unexpected channel-memory request", http.StatusNotFound) + return + } + var payload fakeChannelMemoryIngest + if err := json.NewDecoder(r.Body).Decode(&payload); err != nil { + http.Error(w, "invalid json", http.StatusBadRequest) + return + } + key := payload.ChannelID + "/" + payload.Message.ID + memoryMu.Lock() + ingested[key] = payload + memoryMu.Unlock() + w.WriteHeader(http.StatusAccepted) + _, _ = io.WriteString(w, `{"ok":true}`) + }) + memoryTS := httptest.NewUnstartedServer(memoryHandler) + memoryTS.Listener.Close() + memoryLn, err := net.Listen("tcp", "0.0.0.0:0") + if err != nil { + t.Fatalf("listen channel-memory 0.0.0.0: %v", err) + } + memoryTS.Listener = memoryLn + memoryTS.Start() + defer memoryTS.Close() + _, memoryPort, err := net.SplitHostPort(memoryLn.Addr().String()) + if err != nil { + t.Fatalf("split channel-memory host port: %v", err) + } + // ── Allocate a host port for the wall container ───────────────────── wallPortL, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { @@ -110,6 +145,7 @@ func TestSpikeChannelBackfill(t *testing.T) { "-e", "CLAW_WALL_POLL_INTERVAL=3600", // suppress forward poll noise during the test "-e", "CLAW_WALL_LIMIT=5000", "-e", "CLAW_WALL_ADDR=:8080", + "-e", "CLAW_WALL_CHANNEL_MEMORY_INGEST_URL=http://host.docker.internal:" + memoryPort + "/ingest", imageTag, } if out, err := exec.Command("docker", runArgs...).CombinedOutput(); err != nil { @@ -162,6 +198,30 @@ func TestSpikeChannelBackfill(t *testing.T) { if !strings.Contains(body, "buffer_range=") { t.Fatalf("missing buffer_range in header:\n%s", body) } + + if err := waitForCondition(ctx, func() bool { + memoryMu.Lock() + defer memoryMu.Unlock() + return len(ingested) >= expectedInWindow + }); err != nil { + memoryMu.Lock() + count := len(ingested) + memoryMu.Unlock() + t.Fatalf("channel-memory ingest count never reached %d, got %d: %v", expectedInWindow, count, err) + } + memoryMu.Lock() + newestIngest := ingested[channelID+"/1000000000000359"] + memoryCount := len(ingested) + memoryMu.Unlock() + if memoryCount != expectedInWindow { + t.Fatalf("expected %d unique channel-memory ingests, got %d", expectedInWindow, memoryCount) + } + if newestIngest.Message.ContentHash == "" || !strings.HasPrefix(newestIngest.Message.ContentHash, "sha256:") { + t.Fatalf("expected newest ingest to carry content hash, got %+v", newestIngest) + } + if newestIngest.Scope != "channel:"+channelID || newestIngest.Source.Service != "claw-wall" || newestIngest.Source.Surface != "discord" { + t.Fatalf("unexpected newest ingest scope/source: %+v", newestIngest) + } } type fakeDiscordMessage struct { @@ -177,6 +237,23 @@ type fakeDiscordAuthor struct { GlobalName string `json:"global_name,omitempty"` } +type fakeChannelMemoryIngest struct { + ChannelID string `json:"channel_id"` + Message struct { + ID string `json:"id"` + AuthorName string `json:"author_name"` + CreatedAt string `json:"created_at"` + Content string `json:"content"` + ContentHash string `json:"content_hash"` + } `json:"message"` + Source struct { + Kind string `json:"kind"` + Service string `json:"service"` + Surface string `json:"surface"` + } `json:"source"` + Scope string `json:"scope"` +} + func makeFakeDiscordMessages(channelID string, start time.Time, count int, step time.Duration) []fakeDiscordMessage { out := make([]fakeDiscordMessage, 0, count) for i := 0; i < count; i++ { @@ -312,3 +389,24 @@ func waitForBackfillComplete(ctx context.Context, url string) (string, error) { time.Sleep(250 * time.Millisecond) } } + +func waitForCondition(ctx context.Context, ok func() bool) error { + deadline, _ := ctx.Deadline() + for { + if ok() { + return nil + } + select { + case <-ctx.Done(): + return ctx.Err() + default: + } + if time.Now().After(deadline.Add(-200 * time.Millisecond)) { + if err := ctx.Err(); err != nil { + return err + } + return fmt.Errorf("deadline") + } + time.Sleep(250 * time.Millisecond) + } +} diff --git a/examples/channel-memory/.claw-describe.json b/examples/channel-memory/.claw-describe.json index c3349629..07cab9be 100644 --- a/examples/channel-memory/.claw-describe.json +++ b/examples/channel-memory/.claw-describe.json @@ -1,6 +1,10 @@ { "version": 2, "description": "Channel-memory adapter for source-backed Discord channel awareness digests", + "auth": { + "type": "bearer", + "env": "CHANNEL_MEMORY_TOKEN" + }, "endpoints": [ { "method": "POST", diff --git a/examples/channel-memory/README.md b/examples/channel-memory/README.md index ddd6e8e2..8a756d6d 100644 --- a/examples/channel-memory/README.md +++ b/examples/channel-memory/README.md @@ -36,6 +36,8 @@ LLM worker tracked separately. State is stored under `CHANNEL_MEMORY_DIR` and defaults to `/data/channel-memory`. Set `CHANNEL_MEMORY_DB` to choose an exact SQLite file. +Set `CHANNEL_MEMORY_TOKEN` to require bearer authentication on all data +endpoints. `/health` remains unauthenticated. The schema includes: @@ -57,3 +59,23 @@ uses the repository Go module: ```sh docker build -f examples/channel-memory/Dockerfile -t channel-memory:latest . ``` + +## Pod Wiring + +Declare the adapter once at pod level. `claw up` injects the ingest URL and a +bearer token into `claw-wall`, and injects the matching `CHANNEL_MEMORY_TOKEN` +into the adapter service. + +```yaml +x-claw: + channel-memory: + service: channel-memory + +services: + channel-memory: + build: + context: . + dockerfile: examples/channel-memory/Dockerfile + expose: + - "8080" +``` diff --git a/examples/channel-memory/main.go b/examples/channel-memory/main.go index e093457a..2f384294 100644 --- a/examples/channel-memory/main.go +++ b/examples/channel-memory/main.go @@ -214,6 +214,10 @@ type storedSourceMessage struct { IsCurrent bool } +type handlerConfig struct { + token string +} + func main() { store, err := openStoreFromEnv() if err != nil { @@ -229,7 +233,7 @@ func main() { } log.Printf("channel-memory listening on %s", addr) - if err := http.ListenAndServe(addr, newHandler(store)); err != nil { + if err := http.ListenAndServe(addr, newHandler(store, handlerConfig{token: strings.TrimSpace(os.Getenv("CHANNEL_MEMORY_TOKEN"))})); err != nil { log.Fatalf("listen: %v", err) } } @@ -363,12 +367,19 @@ func (s *channelMemoryStore) initSchema(ctx context.Context) error { return nil } -func newHandler(store *channelMemoryStore) http.Handler { +func newHandler(store *channelMemoryStore, cfgs ...handlerConfig) http.Handler { + cfg := handlerConfig{} + if len(cfgs) > 0 { + cfg = cfgs[0] + } mux := http.NewServeMux() mux.HandleFunc("/ingest", func(w http.ResponseWriter, r *http.Request) { if !requireMethod(w, r, http.MethodPost) { return } + if !authorizeRequest(w, r, cfg) { + return + } var req ingestRequest if !decodeJSON(w, r, &req) { return @@ -385,6 +396,9 @@ func newHandler(store *channelMemoryStore) http.Handler { if !requireMethod(w, r, http.MethodPost) { return } + if !authorizeRequest(w, r, cfg) { + return + } var req sourceMessagesRequest if !decodeJSON(w, r, &req) { return @@ -401,6 +415,9 @@ func newHandler(store *channelMemoryStore) http.Handler { if !requireMethod(w, r, http.MethodPost) { return } + if !authorizeRequest(w, r, cfg) { + return + } var req digestRequest if !decodeJSON(w, r, &req) { return @@ -417,6 +434,9 @@ func newHandler(store *channelMemoryStore) http.Handler { if !requireMethod(w, r, http.MethodPost) { return } + if !authorizeRequest(w, r, cfg) { + return + } var req coverageGapRequest if !decodeJSON(w, r, &req) { return @@ -433,6 +453,9 @@ func newHandler(store *channelMemoryStore) http.Handler { if !requireMethod(w, r, http.MethodPost) { return } + if !authorizeRequest(w, r, cfg) { + return + } var req forgetRequest if !decodeJSON(w, r, &req) { return @@ -1346,6 +1369,18 @@ func requireMethod(w http.ResponseWriter, r *http.Request, method string) bool { return true } +func authorizeRequest(w http.ResponseWriter, r *http.Request, cfg handlerConfig) bool { + token := strings.TrimSpace(cfg.token) + if token == "" { + return true + } + if strings.TrimSpace(r.Header.Get("Authorization")) != "Bearer "+token { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return false + } + return true +} + func decodeJSON(w http.ResponseWriter, r *http.Request, value any) bool { if err := json.NewDecoder(r.Body).Decode(value); err != nil { http.Error(w, fmt.Sprintf("decode JSON: %v", err), http.StatusBadRequest) diff --git a/examples/channel-memory/main_test.go b/examples/channel-memory/main_test.go index e08ec3ee..8d380e69 100644 --- a/examples/channel-memory/main_test.go +++ b/examples/channel-memory/main_test.go @@ -258,6 +258,61 @@ func TestChannelMemoryHTTPAPI(t *testing.T) { } } +func TestChannelMemoryHTTPAuthWhenConfigured(t *testing.T) { + store := newTestStore(t) + defer store.Close() + server := httptest.NewServer(newHandler(store, handlerConfig{token: "memory-token"})) + defer server.Close() + + resp, err := http.Post(server.URL+"/ingest", "application/json", strings.NewReader(`{}`)) + if err != nil { + t.Fatalf("unauthorized post: %v", err) + } + resp.Body.Close() + if resp.StatusCode != http.StatusUnauthorized { + t.Fatalf("expected 401, got %d", resp.StatusCode) + } + + reqBody, err := json.Marshal(ingestRequest{ + ChannelID: "chan-auth", + Message: ingestMessage{ + ID: "401", + AuthorName: "alice", + CreatedAt: "2026-05-21T16:00:00Z", + Content: "authorized content", + ContentHash: "sha256:auth", + }, + }) + if err != nil { + t.Fatalf("marshal authorized request: %v", err) + } + req, err := http.NewRequest(http.MethodPost, server.URL+"/ingest", bytes.NewReader(reqBody)) + if err != nil { + t.Fatalf("authorized request: %v", err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Authorization", "Bearer memory-token") + resp, err = http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("authorized post: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusAccepted { + var got bytes.Buffer + _, _ = got.ReadFrom(resp.Body) + t.Fatalf("expected 202, got %d body=%s", resp.StatusCode, got.String()) + } + + health, err := http.Get(server.URL + "/health") + if err != nil { + t.Fatalf("health: %v", err) + } + defer health.Body.Close() + if health.StatusCode != http.StatusOK { + t.Fatalf("expected health to remain unauthenticated, got %d", health.StatusCode) + } +} + func newTestStore(t *testing.T) *channelMemoryStore { t.Helper() store, err := openStore(filepath.Join(t.TempDir(), "channel-memory.sqlite")) diff --git a/internal/pod/parser.go b/internal/pod/parser.go index b887b06c..7aaa0a16 100644 --- a/internal/pod/parser.go +++ b/internal/pod/parser.go @@ -28,6 +28,7 @@ type rawPodClaw struct { SequentialConformance bool `yaml:"sequential-conformance"` HandlesDefaults map[string]interface{} `yaml:"handles-defaults"` Context *rawContextConfig `yaml:"context"` + ChannelMemory *rawChannelMemoryEntry `yaml:"channel-memory"` Principals []rawPrincipalEntry `yaml:"principals"` AlertWebhooks []string `yaml:"alert-webhooks"` AlertMentions []string `yaml:"alert-mentions"` @@ -122,6 +123,10 @@ type rawMemoryEntry struct { TimeoutMS *int `yaml:"timeout-ms"` } +type rawChannelMemoryEntry struct { + Service string `yaml:"service"` +} + type rawIncludeEntry struct { ID string `yaml:"id"` File string `yaml:"file"` @@ -329,6 +334,12 @@ func Parse(r io.Reader) (*Pod, error) { pod.Services[name] = service } + channelMemory, err := parseChannelMemory(raw.XClaw.ChannelMemory, pod.Services) + if err != nil { + return nil, fmt.Errorf("x-claw.channel-memory: %w", err) + } + pod.ChannelMemory = channelMemory + if pod.Master != "" { svc, ok := pod.Services[pod.Master] if !ok { @@ -634,6 +645,23 @@ func parseMemory(raw *rawMemoryEntry) (*MemoryEntry, error) { }, nil } +func parseChannelMemory(raw *rawChannelMemoryEntry, services map[string]*Service) (*ChannelMemoryConfig, error) { + if raw == nil { + return nil, nil + } + service := strings.TrimSpace(raw.Service) + if service == "" { + return nil, fmt.Errorf("service is required") + } + if service == "claw-wall" { + return nil, fmt.Errorf("service %q is reserved for the conversation wall sidecar", service) + } + if _, ok := services[service]; !ok { + return nil, fmt.Errorf("service %q does not exist", service) + } + return &ChannelMemoryConfig{Service: service}, nil +} + func parseContextConfig(raw *rawContextConfig) (*ContextConfig, error) { if raw == nil { return nil, nil diff --git a/internal/pod/parser_capabilities_test.go b/internal/pod/parser_capabilities_test.go index 601d038c..45362893 100644 --- a/internal/pod/parser_capabilities_test.go +++ b/internal/pod/parser_capabilities_test.go @@ -58,6 +58,33 @@ services: } } +func TestParsePodExtractsChannelMemory(t *testing.T) { + const yaml = ` +x-claw: + pod: channel-pod + channel-memory: + service: channel-memory + +services: + channel-memory: + image: channel-memory:latest + expose: + - "8080" + analyst: + image: analyst:latest + x-claw: + agent: ./AGENTS.md +` + + pod, err := Parse(strings.NewReader(yaml)) + if err != nil { + t.Fatalf("Parse: %v", err) + } + if pod.ChannelMemory == nil || pod.ChannelMemory.Service != "channel-memory" { + t.Fatalf("expected channel-memory service, got %+v", pod.ChannelMemory) + } +} + func TestParsePodDefaultsInheritAndReplaceCapabilityConfig(t *testing.T) { const yaml = ` x-claw: @@ -177,6 +204,20 @@ services: agent: ./AGENTS.md tools: - ... +`, + }, + { + name: "channel memory unknown service", + yaml: ` +x-claw: + pod: invalid-pod + channel-memory: + service: missing-memory +services: + analyst: + image: analyst:latest + x-claw: + agent: ./AGENTS.md `, }, } diff --git a/internal/pod/types.go b/internal/pod/types.go index 1ee9da64..08fb1956 100644 --- a/internal/pod/types.go +++ b/internal/pod/types.go @@ -15,6 +15,7 @@ type Pod struct { ClawAPI *ClawAPIConfig Clawdash *ClawdashConfig // runtime-only dashboard sidecar config, injected by claw up Context *ContextConfig + ChannelMemory *ChannelMemoryConfig Principals []PodPrincipal AlertWebhooks []string // pod-scoped Discord webhook URLs for pool-transition alerts AlertMentions []string // pod-scoped @-mention targets for alerts (e.g. "@wojtek", "@infra") @@ -106,6 +107,10 @@ type MemoryEntry struct { TimeoutMS int } +type ChannelMemoryConfig struct { + Service string +} + type ContextConfig struct { Channel *ChannelContextConfig }