Expose stable source handles in channel awareness

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 {

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 {

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

cmd/claw/spike_rollcall_test.go

@@ -319,13 +319,15 @@ 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() {
 				teardownOnce.Do(func() {
 					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)

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