test(a2a_proxy): #1675 regression — canvas-user callerID must propagate to activity_logs source_id

Adds TestProxyA2A_PollMode_CanvasUserCallerID_PropagatesToActivityLog
which pins the contract broken in #1675: when a canvas-user identity
workspace sends a message to a poll-mode target, the resulting
activity_logs row MUST have source_id = canvas user's workspace UUID
(not NULL).

The test exercises the admin-token canvas-user path (molecli /
break-glass), complementing the existing
TestProxyA2A_PollMode_CanvasUserWithVerifiedSession which covers the
verified-session cookie path.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

workspace-server/internal/handlers/a2a_proxy_test.go

@@ -2444,6 +2444,94 @@ func TestProxyA2A_PollMode_CanvasUserWithVerifiedSession(t *testing.T) {
 	}
 }
 
+// TestProxyA2A_PollMode_CanvasUserCallerID_PropagatesToActivityLog pins
+// the specific contract that broke in molecule-core#1675 (2026-05-22):
+// canvas chat messages from a user with an identity workspace (RFC#637
+// canvas-user-identity rollout) MUST write an activity_logs row whose
+// source_id matches the canvas user's workspace UUID, NOT NULL so the
+// channel plugin's poll path can deliver them as `<channel kind="canvas_user">`
+// tags to the bound Claude Code session, AND the canvas chat-history can
+// re-render the user's own message on reopen.
+//
+// The sibling test TestProxyA2A_PollMode_CanvasUserWithVerifiedSession
+// covers the verified-session cookie path. THIS test covers the admin-token
+// path (molecli / break-glass) which also classifies as canvas-user and
+// bypasses CanCommunicate.
+func TestProxyA2A_PollMode_CanvasUserCallerID_PropagatesToActivityLog(t *testing.T) {
+	mock := setupTestDB(t)
+	setupTestRedis(t)
+	broadcaster := newTestBroadcaster()
+	handler := NewWorkspaceHandler(broadcaster, nil, "http://localhost:8080", t.TempDir())
+
+	const targetWS = "ws-canvas-target-1675"
+	const canvasUserWS = "344a2623-50bf-4ab9-9732-220779305c8f" // shape from #1675 evidence
+
+	// isGenuineCanvasUser checks ADMIN_TOKEN first, so HasAnyLiveToken is
+	// never reached. No SELECT COUNT(*) expectation needed.
+	expectBudgetCheck(mock, targetWS)
+	mock.ExpectQuery("SELECT delivery_mode FROM workspaces WHERE id").
+		WithArgs(targetWS).
+		WillReturnRows(sqlmock.NewRows([]string{"delivery_mode"}).AddRow("poll"))
+
+	// logA2AReceiveQueued looks up the workspace name for the summary.
+	mock.ExpectQuery("SELECT name FROM workspaces WHERE id").
+		WithArgs(targetWS).
+		WillReturnRows(sqlmock.NewRows([]string{"name"}).AddRow("Canvas Target"))
+
+	// CRITICAL: the activity_logs INSERT MUST happen, and its source_id
+	// argument MUST match the canvas user's workspace UUID. The previous
+	// behaviour (sqlmock.ExpectExec with no WithArgs) accepted any args
+	// which is exactly how the regression in #1675 escaped CI: the INSERT
+	// fired, but with source_id=NULL because callerID propagation was
+	// bypassed somewhere upstream. Pin the source_id position explicitly.
+	mock.ExpectExec("INSERT INTO activity_logs").
+		WithArgs(
+			targetWS,                  // workspace_id
+			"a2a_receive",             // activity_type
+			canvasUserWS,              // source_id (NOT NULL the contract this test exists to pin)
+			targetWS,                  // target_id
+			"message/send",            // method
+			sqlmock.AnyArg(),          // summary
+			sqlmock.AnyArg(),          // request_body
+			sqlmock.AnyArg(),          // response_body (nil for queued)
+			sqlmock.AnyArg(),          // tool_trace
+			sqlmock.AnyArg(),          // duration_ms
+			"ok",                      // status
+			sqlmock.AnyArg(),          // error_detail
+		).
+		WillReturnResult(sqlmock.NewResult(0, 1))
+
+	w := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(w)
+	c.Params = gin.Params{{Key: "id", Value: targetWS}}
+	// X-Workspace-ID is the canonical way canvas Next.js identifies the
+	// signed-in user's identity workspace to the platform (per RFC#637).
+	c.Request = httptest.NewRequest("POST", "/workspaces/"+targetWS+"/a2a",
+		bytes.NewBufferString(`{"jsonrpc":"2.0","id":"canvas-1","method":"message/send","params":{"message":{"role":"user","parts":[{"text":"hello from canvas"}]}}}`))
+	c.Request.Header.Set("Content-Type", "application/json")
+	c.Request.Header.Set("X-Workspace-ID", canvasUserWS)
+	c.Request.Header.Set("Authorization", "Bearer test-admin-secret-1675")
+
+	t.Setenv("ADMIN_TOKEN", "test-admin-secret-1675")
+
+	handler.ProxyA2A(c)
+	time.Sleep(50 * time.Millisecond)
+
+	if w.Code != http.StatusOK {
+		t.Fatalf("expected 200 queued, got %d: %s", w.Code, w.Body.String())
+	}
+	var resp map[string]interface{}
+	if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
+		t.Fatalf("response is not valid JSON: %v", err)
+	}
+	if resp["status"] != "queued" {
+		t.Errorf("response.status = %v, want %q", resp["status"], "queued")
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Errorf("unmet sqlmock expectations the activity INSERT may have been skipped OR fired with a different source_id (the #1675 regression shape): %v", err)
+	}
+}
+
 // TestProxyA2A_ForgedSameOrigin_CannotBypassCanCommunicate is the security
 // crux of the #1673 fix and the reason PR #1944 was held. In the combined-
 // tenant SaaS image (CANVAS_PROXY_URL set, CP session verification configured),

workspace-server/internal/handlers/mcp_tools_memory_v2.go

@@ -48,6 +48,7 @@ type memoryV2Deps struct {
 // call. Defining an interface here lets handler tests stub the plugin
 // without spinning up an HTTP server.
 type memoryPluginAPI interface {
+	UpsertNamespace(ctx context.Context, name string, body contract.NamespaceUpsert) (*contract.Namespace, error)
 	CommitMemory(ctx context.Context, namespace string, body contract.MemoryWrite) (*contract.MemoryWriteResponse, error)
 	Search(ctx context.Context, body contract.SearchRequest) (*contract.SearchResponse, error)
 	ForgetMemory(ctx context.Context, id string, body contract.ForgetRequest) error
@@ -117,6 +118,9 @@ func (h *MCPHandler) toolCommitMemoryV2(ctx context.Context, workspaceID string,
 	if !ok {
 		return "", fmt.Errorf("workspace %s cannot write to namespace %s", workspaceID, ns)
 	}
+	if _, err := h.memv2.plugin.UpsertNamespace(ctx, ns, contract.NamespaceUpsert{Kind: kindFromNamespace(ns)}); err != nil {
+		return "", fmt.Errorf("plugin upsert namespace: %w", err)
+	}
 
 	// SAFE-T1201: scrub credential-shaped strings BEFORE the plugin sees
 	// them. Non-negotiable; see memories.go:180.
@@ -171,6 +175,19 @@ func (h *MCPHandler) toolCommitMemoryV2(ctx context.Context, workspaceID string,
 	return string(out), nil
 }
 
+func kindFromNamespace(ns string) contract.NamespaceKind {
+	switch {
+	case strings.HasPrefix(ns, "workspace:"):
+		return contract.NamespaceKindWorkspace
+	case strings.HasPrefix(ns, "team:"):
+		return contract.NamespaceKindTeam
+	case strings.HasPrefix(ns, "org:"):
+		return contract.NamespaceKindOrg
+	default:
+		return contract.NamespaceKindCustom
+	}
+}
+
 // ─────────────────────────────────────────────────────────────────────────────
 // search_memory
 // ─────────────────────────────────────────────────────────────────────────────

workspace-server/internal/handlers/mcp_tools_memory_v2_test.go

@@ -20,11 +20,18 @@ import (
 // --- stubs ---
 
 type stubMemoryPlugin struct {
+	upsertFn func(ctx context.Context, name string, body contract.NamespaceUpsert) (*contract.Namespace, error)
 	commitFn func(ctx context.Context, ns string, body contract.MemoryWrite) (*contract.MemoryWriteResponse, error)
 	searchFn func(ctx context.Context, body contract.SearchRequest) (*contract.SearchResponse, error)
 	forgetFn func(ctx context.Context, id string, body contract.ForgetRequest) error
 }
 
+func (s *stubMemoryPlugin) UpsertNamespace(ctx context.Context, name string, body contract.NamespaceUpsert) (*contract.Namespace, error) {
+	if s.upsertFn != nil {
+		return s.upsertFn(ctx, name, body)
+	}
+	return &contract.Namespace{Name: name, Kind: body.Kind}, nil
+}
 func (s *stubMemoryPlugin) CommitMemory(ctx context.Context, ns string, body contract.MemoryWrite) (*contract.MemoryWriteResponse, error) {
 	if s.commitFn != nil {
 		return s.commitFn(ctx, ns, body)
@@ -159,7 +166,15 @@ func TestMemoryV2Available(t *testing.T) {
 func TestCommitMemoryV2_HappyPathDefaultNamespace(t *testing.T) {
 	db, _, _ := sqlmock.New()
 	defer db.Close()
+	gotUpsertNS := ""
 	h := newV2Handler(t, db, &stubMemoryPlugin{
+		upsertFn: func(_ context.Context, name string, body contract.NamespaceUpsert) (*contract.Namespace, error) {
+			gotUpsertNS = name
+			if body.Kind != contract.NamespaceKindWorkspace {
+				t.Errorf("upsert kind = %q, want workspace", body.Kind)
+			}
+			return &contract.Namespace{Name: name, Kind: body.Kind}, nil
+		},
 		commitFn: func(_ context.Context, ns string, body contract.MemoryWrite) (*contract.MemoryWriteResponse, error) {
 			if ns != "workspace:root-1" {
 				t.Errorf("ns = %q, want default workspace:root-1", ns)
@@ -180,6 +195,9 @@ func TestCommitMemoryV2_HappyPathDefaultNamespace(t *testing.T) {
 	if !strings.Contains(got, `"id":"mem-1"`) {
 		t.Errorf("got = %s", got)
 	}
+	if gotUpsertNS != "workspace:root-1" {
+		t.Errorf("upsert namespace = %q, want workspace:root-1", gotUpsertNS)
+	}
 }
 
 func TestCommitMemoryV2_NamespaceParamUsed(t *testing.T) {

workspace-server/internal/handlers/memories_v2_test.go

@@ -45,6 +45,9 @@ type fakePlugin struct {
 	forgetReq  contract.ForgetRequest
 }
 
+func (f *fakePlugin) UpsertNamespace(ctx context.Context, name string, body contract.NamespaceUpsert) (*contract.Namespace, error) {
+	return &contract.Namespace{Name: name, Kind: body.Kind}, nil
+}
 func (f *fakePlugin) CommitMemory(ctx context.Context, ns string, body contract.MemoryWrite) (*contract.MemoryWriteResponse, error) {
 	return nil, errors.New("not implemented in fake")
 }
@@ -511,11 +514,11 @@ func TestMemoriesV2_Forget_MissingMemoryID_400(t *testing.T) {
 // DisplayName over UUID-prefix fallback (issue #2988).
 func TestNamespaceLabelWithName_PrefersDisplayNameWhenSet(t *testing.T) {
 	cases := []struct {
-		name         string
-		raw          string
-		kind         contract.NamespaceKind
-		display      string
-		want         string
+		name    string
+		raw     string
+		kind    contract.NamespaceKind
+		display string
+		want    string
 	}{
 		{"workspace with name", "workspace:abc-1234", contract.NamespaceKindWorkspace, "mac laptop", "Workspace (mac laptop)"},
 		{"team with name", "team:abc-1234", contract.NamespaceKindTeam, "Engineering", "Team (Engineering)"},
@@ -625,12 +628,12 @@ func TestParseLimit(t *testing.T) {
 	}{
 		{"", memoriesV2DefaultLimit},
 		{"10", 10},
-		{"0", memoriesV2DefaultLimit},        // ≤0 → default, not error
-		{"-5", memoriesV2DefaultLimit},       // negative → default
-		{"abc", memoriesV2DefaultLimit},      // non-numeric → default
-		{"99999", memoriesV2MaxLimit},        // over cap → clamped
-		{"100", memoriesV2MaxLimit},          // exactly cap → kept
-		{"99", 99},                           // just under cap → kept
+		{"0", memoriesV2DefaultLimit},   // ≤0 → default, not error
+		{"-5", memoriesV2DefaultLimit},  // negative → default
+		{"abc", memoriesV2DefaultLimit}, // non-numeric → default
+		{"99999", memoriesV2MaxLimit},   // over cap → clamped
+		{"100", memoriesV2MaxLimit},     // exactly cap → kept
+		{"99", 99},                      // just under cap → kept
 	}
 	for _, tc := range cases {
 		t.Run("raw="+tc.raw, func(t *testing.T) {
@@ -741,11 +744,11 @@ func TestWithMemoryV2_FluentReturnsReceiver(t *testing.T) {
 
 func TestShortID(t *testing.T) {
 	cases := map[string]string{
-		"":                    "",
-		"short":               "short",
-		"exactly8":            "exactly8",
-		"longer-than-eight":   "longer-t",
-		"abc-1234-5678-90ab":  "abc-1234",
+		"":                   "",
+		"short":              "short",
+		"exactly8":           "exactly8",
+		"longer-than-eight":  "longer-t",
+		"abc-1234-5678-90ab": "abc-1234",
 	}
 	for in, want := range cases {
 		if got := shortID(in); got != want {