diff --git a/cmd/claw-wall/main_test.go b/cmd/claw-wall/main_test.go index bcb56de..cbaf06b 100644 --- a/cmd/claw-wall/main_test.go +++ b/cmd/claw-wall/main_test.go @@ -265,6 +265,9 @@ func TestChannelAwarenessHandlerReturnsRawWindow(t *testing.T) { if strings.Contains(text, "older signal") || !strings.Contains(text, "newer signal") { t.Fatalf("expected newest bounded awareness body, got %q", text) } + if !strings.Contains(text, "source=chan-1/101") { + t.Fatalf("expected stable source handle in awareness body, got %q", text) + } } func TestChannelAwarenessHeaderReportsBackfillStatus(t *testing.T) { @@ -330,6 +333,32 @@ func TestToolSearchRequiresAuthAndChannelAllowlist(t *testing.T) { if result.Status != "ok" || len(result.Messages) != 1 || result.Messages[0].ID != "100" { t.Fatalf("unexpected search result: %+v", result) } + if result.Messages[0].SourceHandle != "chan-1/100" { + t.Fatalf("expected source handle in search result, got %+v", result.Messages[0]) + } + + req, err = http.NewRequest(http.MethodPost, server.URL+"/get_channel_messages", strings.NewReader(`{"channels":["chan-1"],"message_ids":["100"]}`)) + if err != nil { + t.Fatalf("request exact message: %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 exact message: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(resp.Body) + t.Fatalf("expected 200 for exact message, got %d: %s", resp.StatusCode, string(body)) + } + var exact retrievalResult + if err := json.NewDecoder(resp.Body).Decode(&exact); err != nil { + t.Fatalf("decode exact response: %v", err) + } + if exact.Status != "ok" || len(exact.Messages) != 1 || exact.Messages[0].SourceHandle != "chan-1/100" { + t.Fatalf("unexpected exact message result: %+v", exact) + } req, err = http.NewRequest(http.MethodPost, server.URL+"/get_channel_messages", strings.NewReader(`{"channels":["chan-2"],"message_ids":["200"]}`)) if err != nil { diff --git a/cmd/claw-wall/store.go b/cmd/claw-wall/store.go index 32c231e..7279d60 100644 --- a/cmd/claw-wall/store.go +++ b/cmd/claw-wall/store.go @@ -29,11 +29,12 @@ const ( ) type wallMessage struct { - ID string `json:"id"` - ChannelID string `json:"channel_id"` - Author string `json:"author"` - Content string `json:"content"` - Timestamp time.Time `json:"timestamp"` + ID string `json:"id"` + ChannelID string `json:"channel_id"` + SourceHandle string `json:"source_handle,omitempty"` + Author string `json:"author"` + Content string `json:"content"` + Timestamp time.Time `json:"timestamp"` } type channelBuffer struct { @@ -187,6 +188,7 @@ func (s *conversationStore) mergeAt(channelID string, messages []wallMessage, no if _, exists := state.seenIDs[msg.ID]; exists { continue } + msg.SourceHandle = stableSourceHandle(msg) state.seenIDs[msg.ID] = struct{}{} state.messages = append(state.messages, msg) } @@ -840,9 +842,21 @@ func formatWallMessages(messages []wallMessage) string { } func formatWallMessage(msg wallMessage) string { + if handle := stableSourceHandle(msg); handle != "" { + return fmt.Sprintf("[%s source=%s] %s: %s", formatWallTimestamp(msg.Timestamp), handle, msg.Author, msg.Content) + } return fmt.Sprintf("[%s] %s: %s", formatWallTimestamp(msg.Timestamp), msg.Author, msg.Content) } +func stableSourceHandle(msg wallMessage) string { + channelID := strings.TrimSpace(msg.ChannelID) + messageID := strings.TrimSpace(msg.ID) + if channelID == "" || messageID == "" { + return "" + } + return channelID + "/" + messageID +} + func contextKindFromRequest(r *http.Request, result tailResult) string { kind := strings.TrimSpace(r.URL.Query().Get("context_kind")) switch kind { diff --git a/cmd/claw/spike_channel_backfill_test.go b/cmd/claw/spike_channel_backfill_test.go index 92510e3..8f1bcec 100644 --- a/cmd/claw/spike_channel_backfill_test.go +++ b/cmd/claw/spike_channel_backfill_test.go @@ -152,6 +152,10 @@ func TestSpikeChannelBackfill(t *testing.T) { if !strings.Contains(body, wantAvailable) { t.Fatalf("expected %q in header, got body:\n%s", wantAvailable, body) } + wantSource := "source=" + channelID + "/1000000000000359" + if !strings.Contains(body, wantSource) { + t.Fatalf("expected newest message source handle %q in body, got body:\n%s", wantSource, body) + } // Buffer range should start near 24h ago (the oldest in-window message // is at now-24h, since messages are 6min apart and index 120 = -24h // exactly). Allow some clock skew. @@ -161,11 +165,11 @@ func TestSpikeChannelBackfill(t *testing.T) { } type fakeDiscordMessage struct { - ID string `json:"id"` - Content string `json:"content"` - Timestamp string `json:"timestamp"` - Author fakeDiscordAuthor `json:"author"` - ChannelID string `json:"channel_id,omitempty"` + ID string `json:"id"` + Content string `json:"content"` + Timestamp string `json:"timestamp"` + Author fakeDiscordAuthor `json:"author"` + ChannelID string `json:"channel_id,omitempty"` } type fakeDiscordAuthor struct { @@ -308,4 +312,3 @@ func waitForBackfillComplete(ctx context.Context, url string) (string, error) { time.Sleep(250 * time.Millisecond) } } - diff --git a/cmd/claw/spike_rollcall_test.go b/cmd/claw/spike_rollcall_test.go index 35b5362..bdc1a8e 100644 --- a/cmd/claw/spike_rollcall_test.go +++ b/cmd/claw/spike_rollcall_test.go @@ -319,6 +319,7 @@ func TestSpikeRollCall(t *testing.T) { agentContainerID := rollcallResolveContainerID(t, generatedPath, agent.name) cllamaContainerID := rollcallResolveContainerID(t, generatedPath, "cllama") clawdashContainerID := rollcallResolveContainerID(t, generatedPath, "clawdash") + clawWallContainerID := rollcallResolveContainerID(t, generatedPath, "claw-wall") var teardownOnce sync.Once teardown := func() { @@ -326,6 +327,7 @@ func TestSpikeRollCall(t *testing.T) { rollcallLogContainer(t, agentContainerID) rollcallLogContainer(t, cllamaContainerID) rollcallLogContainer(t, clawdashContainerID) + rollcallLogContainer(t, clawWallContainerID) spikeCleanupProject(composeProject, generatedPath) _ = os.Remove(generatedPath) _ = os.RemoveAll(runtimeDir) @@ -342,6 +344,7 @@ func TestSpikeRollCall(t *testing.T) { }) spikeWaitHealthy(t, agentContainerID, 120*time.Second) + spikeWaitHealthy(t, clawWallContainerID, 60*time.Second) auditWindowStart := time.Now() triggerMsg := fmt.Sprintf("<@%s> Runtime check: introduce yourself and state what runtime you are running on.", botID) @@ -357,6 +360,7 @@ func TestSpikeRollCall(t *testing.T) { 2*time.Minute, ) t.Logf("found %s response: %q", agent.runtime, rollcallTruncate(response, 120)) + spikeVerifyContainerChannelAwarenessSourceHandle(t, clawWallContainerID, channelID, 60*time.Second) rollcallAssertAuditTelemetry(t, podPath, agent.name, agent.runtime, auditWindowStart) rollcallAssertSessionHistory(t, sessionHistoryDir, agent.name) diff --git a/cmd/claw/spike_test.go b/cmd/claw/spike_test.go index 0e057ff..019c0ea 100644 --- a/cmd/claw/spike_test.go +++ b/cmd/claw/spike_test.go @@ -691,6 +691,7 @@ func TestSpikeComposeUp(t *testing.T) { // trading-api posts its own startup message to Discord via webhook — this // proves non-claw services receive env vars (DISCORD_TRADING_API_WEBHOOK). spikeVerifyDiscordGreeting(t, env["TIVERTON_BOT_TOKEN"], channelID, "trading-api online", 15*time.Second) + spikeVerifyChannelAwarenessSourceHandle(t, channelID, 60*time.Second) // The startup message must contain Discord mentions for openclaw agents. // CLAW_HANDLE_* vars are broadcast to all pod services by claw, so trading-api @@ -1063,6 +1064,29 @@ func spikeVerifyDiscordGreeting(t *testing.T, botToken, channelID, expectedSubst t.Errorf("Discord greeting %q not found in channel %s after %v", expectedSubstr, channelID, timeout) } +func spikeVerifyChannelAwarenessSourceHandle(t *testing.T, channelID string, timeout time.Duration) { + t.Helper() + spikeVerifyContainerChannelAwarenessSourceHandle(t, spikeContainerName("claw-wall"), channelID, timeout) +} + +func spikeVerifyContainerChannelAwarenessSourceHandle(t *testing.T, containerName, channelID string, timeout time.Duration) { + t.Helper() + url := fmt.Sprintf("http://127.0.0.1:8080/channel-awareness?channels=%s&since=24h&limit=50&max_chars=200000", channelID) + want := "source=" + channelID + "/" + deadline := time.Now().Add(timeout) + var lastBody string + for time.Now().Before(deadline) { + out, err := exec.Command("docker", "exec", containerName, "wget", "-qO-", url).CombinedOutput() + lastBody = string(out) + if err == nil && strings.Contains(lastBody, want) { + t.Logf("found channel-awareness source handle for channel %s", channelID) + return + } + time.Sleep(3 * time.Second) + } + t.Errorf("channel-awareness source handle %q not found after %v; last body:\n%s", want, timeout, lastBody) +} + func spikeImageExists(tag string) bool { out, err := exec.Command("docker", "image", "inspect", "--format", "{{.Id}}", tag).Output() return err == nil && len(strings.TrimSpace(string(out))) > 0