feat(activity): flat-upload-manifest arm in extractAttachmentsFromRequestBody

Adds a second walk arm for the canvas chat_upload_receive shape: a flat upload manifest at request_body root (no JSON-RPC envelope) with camelCase mimeType. Normalizes to snake_case mime_type on emit + derives kind from the mime prefix (image/* -> image, audio/* -> audio, video/* -> video, else -> file). Empirical surface: 2026-05-21 ~23:12Z canvas-user pasted a PNG, the activity row's request_body was {uri, name, size, file_id, mimeType} with no params/message/parts wrapper, and ?include=peer_info projected attachments: null instead of the expected one-element array. The new arm handles this shape uniformly so every downstream adaptor (channel / telegram / codex / hermes) sees a populated attachments[] with zero per-adaptor parsing. Per the three-layer data-responsibility rule (platform / base / adaptor), upload-shape parsing belongs at Layer 1 (the platform's projection), not in adaptors. Tests: - TestKindFromMimeType (12 cases) pins the mime->kind derivation. - TestExtractAttachmentsFromRequestBody_FlatUpload_* (8 sub-tests) cover image / audio / video / generic-file / no-mime-fallback / snake-case mime_type accepted / file_id-only-skipped / name-only-kept. - TestExtractAttachmentsFromRequestBody_MessagePartsTakesPrecedenceOverFlat pins that a pathological body with BOTH shapes uses the parts[] arm (the documented inbound, historically the only one extracted). - TestActivityList_IncludePeerInfo_ChatUploadReceiveCanvasRow is a wire- level integration test against the empirical 2026-05-21 row shape. Follow-up: workspace-runtime#37's _extract_attachments_from_request_body gets the same flat-upload arm for pre-Layer-1 platform parity. Not required for already-L1-enabled platforms (which read the row-level attachments[] field this projection populates). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Use literal region for AWS secrets janitor (#1655 )
2026-05-21 16:26:01 -07:00 · 2026-05-21 21:52:07 +00:00 · 2026-05-21 14:42:23 -07:00 · 2026-05-21 21:41:18 +00:00 · 2026-05-21 20:56:00 +00:00 · 2026-05-21 13:50:18 -07:00
7 changed files with 1115 additions and 67 deletions
@@ -86,6 +86,7 @@ on:
      - 'workspace-server/internal/handlers/registry.go'
      - 'workspace-server/internal/handlers/workspace.go'
      - 'tests/e2e/test_peer_visibility_mcp_staging.sh'
+      - 'tests/e2e/test_peer_visibility_token_mint_staging.sh'
      - 'tests/e2e/test_peer_visibility_mcp_local.sh'
      - 'tests/e2e/lib/peer_visibility_assert.sh'
      - '.gitea/workflows/e2e-peer-visibility.yml'
@@ -98,6 +99,7 @@ on:
      - 'workspace-server/internal/handlers/registry.go'
      - 'workspace-server/internal/handlers/workspace.go'
      - 'tests/e2e/test_peer_visibility_mcp_staging.sh'
+      - 'tests/e2e/test_peer_visibility_token_mint_staging.sh'
      - 'tests/e2e/test_peer_visibility_mcp_local.sh'
      - 'tests/e2e/lib/peer_visibility_assert.sh'
      - '.gitea/workflows/e2e-peer-visibility.yml'
@@ -137,8 +139,14 @@ jobs:
          echo "lib/peer_visibility_assert.sh — bash syntax OK"
          bash -n tests/e2e/test_peer_visibility_mcp_staging.sh
          echo "test_peer_visibility_mcp_staging.sh — bash syntax OK"
+          bash -n tests/e2e/test_peer_visibility_token_mint_staging.sh
+          echo "test_peer_visibility_token_mint_staging.sh — bash syntax OK"
          bash -n tests/e2e/test_peer_visibility_mcp_local.sh
          echo "test_peer_visibility_mcp_local.sh — bash syntax OK"
+          if rg -n '/admin/workspaces/.*/test-token|test-token' tests/e2e/test_*staging*.sh; then
+            echo "::error::staging E2E must not use dev-only /admin/workspaces/:id/test-token; use production-safe admin token minting instead"
+            exit 1
+          fi
          echo "Staging fresh-provision MCP list_peers E2E runs on push to"
          echo "main / workflow_dispatch / daily cron (30+ min EC2 boot)."
          echo "The LOCAL backend runs in the peer-visibility-local job"
@@ -29,7 +29,8 @@ name: publish-workspace-server-image
 # Optional staging tenant mirror target:
 #   004947743811.dkr.ecr.us-east-2.amazonaws.com/molecule-ai/platform-tenant
 # Required secrets: AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AUTO_SYNC_TOKEN
-# Optional secrets: AWS_STAGING_ECR_ACCESS_KEY_ID, AWS_STAGING_ECR_SECRET_ACCESS_KEY
+# Staging ECR grants the primary SSOT-managed publisher principal repository
+# policy access, so no persistent staging AWS access keys are required.
 #
 # mc#711: Docker daemon not accessible on ubuntu-latest runner (molecule-canonical-1
 # shows client-only in `docker info` — daemon not running). DinD mount is present but
@@ -186,9 +187,10 @@ jobs:
            --push .

      # Build + push tenant image (Go platform + Next.js canvas in one image).
-      # When staging ECR publisher credentials are configured, push the same
-      # build to the staging account too so fresh staging/E2E tenants can pull
-      # without cross-account ECR permissions.
+      # Push the same build to the staging account too so fresh staging/E2E
+      # tenants can pull without cross-account ECR reads. The staging ECR repo
+      # policy trusts the primary SSOT-managed publisher principal; do not add
+      # separate persistent staging AWS access keys here.
      - name: Build & push tenant image to ECR (staging-<sha> + staging-latest)
        env:
          TENANT_IMAGE_NAME: ${{ env.TENANT_IMAGE_NAME }}
@@ -199,32 +201,22 @@ jobs:
          REPO: ${{ github.repository }}
          AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
          AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
-          AWS_STAGING_ECR_ACCESS_KEY_ID: ${{ secrets.AWS_STAGING_ECR_ACCESS_KEY_ID }}
-          AWS_STAGING_ECR_SECRET_ACCESS_KEY: ${{ secrets.AWS_STAGING_ECR_SECRET_ACCESS_KEY }}
          AWS_DEFAULT_REGION: us-east-2
        run: |
          set -euo pipefail
          ECR_REGISTRY="${TENANT_IMAGE_NAME%%/*}"
+          STAGING_ECR_REGISTRY="${STAGING_TENANT_IMAGE_NAME%%/*}"
          aws ecr get-login-password --region us-east-2 | \
            docker login --username AWS --password-stdin "${ECR_REGISTRY}"
+          aws ecr get-login-password --region us-east-2 | \
+            docker login --username AWS --password-stdin "${STAGING_ECR_REGISTRY}"

          build_tags=(
            --tag "${TENANT_IMAGE_NAME}:${TAG_SHA}"
            --tag "${TENANT_IMAGE_NAME}:${TAG_LATEST}"
+            --tag "${STAGING_TENANT_IMAGE_NAME}:${TAG_SHA}"
+            --tag "${STAGING_TENANT_IMAGE_NAME}:${TAG_LATEST}"
          )
-          if [ -n "${AWS_STAGING_ECR_ACCESS_KEY_ID:-}" ] && [ -n "${AWS_STAGING_ECR_SECRET_ACCESS_KEY:-}" ]; then
-            STAGING_ECR_REGISTRY="${STAGING_TENANT_IMAGE_NAME%%/*}"
-            AWS_ACCESS_KEY_ID="${AWS_STAGING_ECR_ACCESS_KEY_ID}" \
-            AWS_SECRET_ACCESS_KEY="${AWS_STAGING_ECR_SECRET_ACCESS_KEY}" \
-              aws ecr get-login-password --region us-east-2 | \
-              docker login --username AWS --password-stdin "${STAGING_ECR_REGISTRY}"
-            build_tags+=(
-              --tag "${STAGING_TENANT_IMAGE_NAME}:${TAG_SHA}"
-              --tag "${STAGING_TENANT_IMAGE_NAME}:${TAG_LATEST}"
-            )
-          else
-            echo "::notice::Skipping staging ECR tenant push; AWS_STAGING_ECR_ACCESS_KEY_ID/AWS_STAGING_ECR_SECRET_ACCESS_KEY are not configured."
-          fi

          docker buildx build \
            --file ./workspace-server/Dockerfile.tenant \
@@ -40,14 +40,12 @@ name: Sweep stale AWS Secrets Manager secrets
 # the mostly-orphan tunnels) refuses to nuke past the threshold.

 on:
-  # Disabled as an hourly schedule until the dedicated
-  # AWS_SECRETS_JANITOR_* key exists in the key-management SSOT and is
-  # mirrored into Gitea. Falling back to the molecule-cp app principal is
-  # intentionally not allowed: it lacks account-wide ListSecrets, and
-  # granting that to an application credential would weaken least privilege.
-  #
-  # Keep the manual trigger so operators can validate the workflow immediately
-  # after provisioning the janitor key, then restore the hourly :30 schedule.
+  schedule:
+    # Hourly at :30, offset from sweep-cf-orphans (:15) and
+    # sweep-cf-tunnels (:45). This janitor is intentionally schedule-only
+    # for deletes; manual dispatch is forced to dry-run below because Gitea
+    # 1.22.6 rejects workflow_dispatch.inputs.
+    - cron: '30 * * * *'
  workflow_dispatch:
 # Don't let two sweeps race the same AWS account.
 concurrency:
@@ -64,22 +62,24 @@ jobs:
  sweep:
    name: Sweep AWS Secrets Manager
    runs-on: ubuntu-latest
-    # Phase 3 (RFC #219 §1): surface broken workflows without blocking.
-    # mc#774: pre-existing continue-on-error mask; root-fix and remove, do not renew silently.
-    continue-on-error: true
+    # This is a cost/leak janitor. A scheduled failure must be red so
+    # operators know tenant bootstrap secrets may be leaking.
    # 30 min cap, mirroring the other janitors. AWS DeleteSecret is
    # fast (~0.3s/call) so even a 100+ backlog drains in seconds
    # under the 8-way xargs parallelism, but the cap is set generously
    # to leave headroom for any actual API hang.
    timeout-minutes: 30
    env:
-      AWS_REGION: ${{ secrets.AWS_REGION || 'us-east-1' }}
+      # Keep this literal. Gitea/act_runner 1.22.6 can mis-render
+      # secret-backed expressions with `||`, which produced an invalid
+      # Secrets Manager endpoint in the scheduled janitor.
+      AWS_REGION: us-east-2
      AWS_ACCESS_KEY_ID: ${{ secrets.AWS_SECRETS_JANITOR_ACCESS_KEY_ID }}
      AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRETS_JANITOR_SECRET_ACCESS_KEY }}
      CP_ADMIN_API_TOKEN: ${{ secrets.CP_ADMIN_API_TOKEN }}
      CP_STAGING_ADMIN_API_TOKEN: ${{ secrets.CP_STAGING_ADMIN_API_TOKEN }}
-      MAX_DELETE_PCT: ${{ github.event.inputs.max_delete_pct || '50' }}
-      GRACE_HOURS: ${{ github.event.inputs.grace_hours || '24' }}
+      MAX_DELETE_PCT: 50
+      GRACE_HOURS: 24

    steps:
      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
@@ -114,17 +114,25 @@ jobs:

      - name: Run sweep
        if: steps.verify.outputs.skip != 'true'
-        # Schedule-vs-dispatch dry-run asymmetry mirrors sweep-cf-tunnels:
-        #   - Scheduled: input empty → "false" → --execute (the whole
-        #     point of an hourly janitor).
-        #   - Manual workflow_dispatch: input default true → dry-run;
-        #     operator must flip it to actually delete.
+        # Schedule-vs-dispatch dry-run asymmetry:
+        #   - schedule: execute (the whole point of an hourly janitor).
+        #   - workflow_dispatch: dry-run. Gitea 1.22.6 rejects
+        #     workflow_dispatch.inputs, so there is no safe manual
+        #     "flip it to execute" toggle in this workflow.
+        # The script's MAX_DELETE_PCT gate (default 50%) remains the
+        # second line of defense regardless of trigger.
        run: |
          set -euo pipefail
-          if [ "${{ github.event.inputs.dry_run || 'false' }}" = "true" ]; then
+          if [ "${{ github.event_name }}" = "workflow_dispatch" ]; then
            echo "Running in dry-run mode — no deletions"
            bash scripts/ops/sweep-aws-secrets.sh
          else
            echo "Running with --execute — will delete identified orphans"
            bash scripts/ops/sweep-aws-secrets.sh --execute
          fi
+
+      - name: Notify on sweep failure
+        if: failure()
+        run: |
+          echo "::error::sweep-aws-secrets FAILED — AWS tenant bootstrap secrets may be leaking. Check missing Gitea secrets, staging/prod CP admin tokens, AWS janitor IAM permissions, or the script safety gate."
+          exit 1
@@ -54,6 +54,9 @@
 #   E2E_PROVISION_TIMEOUT_SECS  default 1800 (hermes/openclaw cold EC2 budget)
 #   E2E_MINIMAX_API_KEY / E2E_ANTHROPIC_API_KEY / E2E_OPENAI_API_KEY
 #                          LLM provider key injected so the runtime can boot
+#   PV_TOKEN_DIAGNOSTIC_ONLY
+#                          1 -> stop after create/token acquisition. Useful
+#                          to classify Hermes-only vs shared auth-route issues.
 #   E2E_KEEP_ORG           1 → skip teardown (local debugging only)
 #
 # Exit codes:
@@ -232,6 +235,12 @@ for i in $(seq 1 120); do
  curl -fsS "$TENANT_URL/health" -m 5 -k >/dev/null 2>&1 && { log "    /health ok (attempt $i)"; break; }
  sleep 5
 done
+BUILDINFO=$(curl -fsS "$TENANT_URL/buildinfo" -m 10 2>/dev/null || true)
+if [ -n "$BUILDINFO" ]; then
+  log "    tenant buildinfo: $(echo "$BUILDINFO" | head -c 300)"
+else
+  log "    tenant buildinfo unavailable"
+fi

 # ─── 4. Provision the parent + one sibling per runtime under test ──────
 # Inject the LLM provider key so each runtime can authenticate at boot.
@@ -256,6 +265,8 @@ log "    PARENT_ID=$PARENT_ID"
 # WS_IDS[runtime]=id ; WS_TOKENS[runtime]=auth_token (the MCP bearer)
 declare -A WS_IDS WS_TOKENS
 ALL_WS_IDS="$PARENT_ID"
+TOKEN_ERRORS=0
+TOKEN_ERROR_SUMMARY=""
 for rt in $PV_RUNTIMES; do
  R=$(tenant_call POST /workspaces \
    -d "{\"name\":\"pv-$rt\",\"runtime\":\"$rt\",\"tier\":2,\"parent_id\":\"$PARENT_ID\",\"secrets\":$SECRETS_JSON}")
@@ -263,7 +274,7 @@ for rt in $PV_RUNTIMES; do
  # External-like runtimes may return connection.auth_token on create.
  # Managed container runtimes usually return only id/status here, then
  # receive their bearer through registry/bootstrap; for this literal MCP
-  # driver we mint an admin test token below.
+  # driver we mint through the production-safe admin token route below.
  WTOK=$(echo "$R" | extract_auth_token)
  [ -n "$WID" ] || fail "$rt workspace create failed: $(echo "$R" | head -c 300)"
  TOKEN_DIAG=""
@@ -275,22 +286,28 @@ for rt in $PV_RUNTIMES; do
    TOKEN_DIAG="POST /admin/workspaces/$WID/tokens -> HTTP $TTOK_CODE body: $(echo "$TTOK_RESP" | redact_token_body)"
    rm -f "$TTOK_FILE"
  fi
-  if [ -z "$WTOK" ]; then
-    TTOK_FILE=$(mktemp)
-    TTOK_CODE=$(tenant_call_capture GET "/admin/workspaces/$WID/test-token" "$TTOK_FILE" 2>/dev/null || echo "curl_error")
-    TTOK_RESP=$(cat "$TTOK_FILE" 2>/dev/null || true)
-    WTOK=$(echo "$TTOK_RESP" | extract_auth_token)
-    TOKEN_DIAG="${TOKEN_DIAG}
-GET /admin/workspaces/$WID/test-token -> HTTP $TTOK_CODE body: $(echo "$TTOK_RESP" | redact_token_body)"
-    rm -f "$TTOK_FILE"
-  fi
-  [ -n "$WTOK" ] || fail "$rt workspace did not return or mint an auth_token — cannot drive its MCP call (create_resp: $(echo "$R" | redact_token_body); token_fallbacks: $TOKEN_DIAG)"
  WS_IDS[$rt]="$WID"
+  if [ -z "$WTOK" ]; then
+    TOKEN_ERRORS=$((TOKEN_ERRORS + 1))
+    TOKEN_ERROR_SUMMARY="${TOKEN_ERROR_SUMMARY}
+[$rt] workspace did not return or mint an auth_token — cannot drive its MCP call (workspace_id=$WID; create_resp: $(echo "$R" | redact_token_body); token_fallbacks: $TOKEN_DIAG)"
+    log "    $rt → $WID (token acquisition failed; continuing to classify other runtimes)"
+    continue
+  fi
  WS_TOKENS[$rt]="$WTOK"
  ALL_WS_IDS="$ALL_WS_IDS $WID"
  log "    $rt → $WID"
 done

+if [ "$TOKEN_ERRORS" -gt 0 ]; then
+  fail "token acquisition failed for $TOKEN_ERRORS runtime(s):$TOKEN_ERROR_SUMMARY"
+fi
+
+if [ "${PV_TOKEN_DIAGNOSTIC_ONLY:-0}" = "1" ]; then
+  ok "token diagnostic passed for runtimes: $PV_RUNTIMES"
+  exit 0
+fi
+
 # ─── 5. Wait for every sibling online ──────────────────────────────────
 log "5/6 waiting for all workspaces status=online (up to ${PROVISION_TIMEOUT_SECS}s — cold boot)..."
 WS_DEADLINE=$(( $(date +%s) + PROVISION_TIMEOUT_SECS ))
@@ -0,0 +1,22 @@
+#!/usr/bin/env bash
+# Staging E2E diagnostic — classify peer-visibility token acquisition.
+#
+# This is intentionally narrower than test_peer_visibility_mcp_staging.sh:
+# it provisions the same throwaway org, creates managed sibling workspaces,
+# and stops immediately after auth_token acquisition. The default runtime set
+# compares hermes with claude-code so a failure is easy to classify:
+#   - hermes fails, claude-code passes -> Hermes/runtime-specific
+#   - both fail                         -> shared admin/auth/proxy route
+#
+# Required env matches test_peer_visibility_mcp_staging.sh:
+#   MOLECULE_ADMIN_TOKEN
+# Optional:
+#   MOLECULE_CP_URL, E2E_RUN_ID, PV_RUNTIMES, E2E_KEEP_ORG,
+#   E2E_MINIMAX_API_KEY / E2E_ANTHROPIC_API_KEY / E2E_OPENAI_API_KEY
+
+set -euo pipefail
+
+export PV_RUNTIMES="${PV_RUNTIMES:-hermes claude-code}"
+export PV_TOKEN_DIAGNOSTIC_ONLY=1
+
+exec "$(dirname "${BASH_SOURCE[0]}")/test_peer_visibility_mcp_staging.sh"
@@ -67,7 +67,213 @@ func NewActivityHandler(b *events.Broadcaster) *ActivityHandler {
 	return &ActivityHandler{broadcaster: b}
 }

-// List handles GET /workspaces/:id/activity?type=&source=&limit=&since_secs=&since_id=
+// extractAttachmentsFromRequestBody walks a JSON-RPC a2a inbound body to
+// surface attachments (file/image/audio/video) as a flat `attachments[]`
+// projection so callers don't have to drill into the request_body shape
+// themselves.
+//
+// Two body shapes are walked in order:
+//
+//  1. a2a-sdk v1 message-part envelope (peer_agent inbound):
+//
+//     {"jsonrpc":"2.0","method":"message/send","params":{
+//        "message":{"parts":[
+//          {"kind":"text", "text":"hi"},
+//          {"kind":"file", "file":{"uri":"workspace:foo.pdf","mime_type":"application/pdf","name":"foo.pdf"}},
+//          {"kind":"image","file":{"uri":"workspace:bar.png","mime_type":"image/png","name":"bar.png"}},
+//        ]}}}
+//
+//  2. canvas chat_upload_receive flat manifest (canvas_user upload):
+//
+//     {"uri":"platform-pending:<ws>/<file>",
+//      "name":"pasted.png",
+//      "size":12345,
+//      "file_id":"<uuid>",
+//      "mimeType":"image/png"}
+//
+//     The canvas upload pipe writes a single manifest directly at the
+//     root of request_body (no JSON-RPC envelope) with camelCase
+//     `mimeType`. We normalize to snake_case `mime_type` on the way out
+//     so every downstream adaptor (channel / telegram / codex / hermes)
+//     sees one wire shape regardless of which inbound shape produced it.
+//
+// Returns nil (omit-from-JSON) when the body has no attachments — the
+// `?include=peer_info` envelope projects this as an array iff non-empty.
+//
+// Defensive on every step: any missing key / wrong-shape value falls
+// through to the next arm or returns nil instead of panicking. The
+// activity_logs row could carry literally any JSON in request_body
+// (legacy formats, future formats); we only commit to the documented
+// shapes and silently skip anything else.
+func extractAttachmentsFromRequestBody(raw []byte) []map[string]interface{} {
+	if len(raw) == 0 {
+		return nil
+	}
+	var body map[string]interface{}
+	if err := json.Unmarshal(raw, &body); err != nil {
+		return nil
+	}
+	if atts := extractAttachmentsFromMessageParts(body); len(atts) > 0 {
+		return atts
+	}
+	if att := extractAttachmentFromFlatUploadManifest(body); att != nil {
+		return []map[string]interface{}{att}
+	}
+	return nil
+}
+
+// extractAttachmentsFromMessageParts handles the a2a-sdk v1 shape:
+// body.params.message.parts[]. Walks file/image/audio parts; honors v1
+// `kind` and v0 `type` discriminators; accepts nested `.file` sub-object
+// or inlined uri/mime_type/name on the part itself.
+func extractAttachmentsFromMessageParts(body map[string]interface{}) []map[string]interface{} {
+	params, ok := body["params"].(map[string]interface{})
+	if !ok {
+		return nil
+	}
+	message, ok := params["message"].(map[string]interface{})
+	if !ok {
+		return nil
+	}
+	parts, ok := message["parts"].([]interface{})
+	if !ok {
+		return nil
+	}
+	out := make([]map[string]interface{}, 0)
+	for _, p := range parts {
+		part, ok := p.(map[string]interface{})
+		if !ok {
+			continue
+		}
+		// a2a-sdk v1 uses "kind"; older v0 callers sent "type". Accept
+		// both for the discriminator — same defensive read pattern as
+		// the runtime-side extract_text helper.
+		kind, _ := part["kind"].(string)
+		if kind == "" {
+			kind, _ = part["type"].(string)
+		}
+		if kind != "file" && kind != "image" && kind != "audio" {
+			continue
+		}
+		// The file sub-object holds uri/mime_type/name. The a2a-sdk v1
+		// shape nests under "file"; some legacy payloads inlined the
+		// fields onto the part itself. Support both.
+		var fileObj map[string]interface{}
+		if f, ok := part["file"].(map[string]interface{}); ok {
+			fileObj = f
+		} else {
+			fileObj = part
+		}
+		uri, _ := fileObj["uri"].(string)
+		mimeType, _ := fileObj["mime_type"].(string)
+		name, _ := fileObj["name"].(string)
+		// At minimum we need either a uri or a name to be useful.
+		// Empty-part entries are skipped (they're a malformed inbound
+		// — surface nothing rather than emit a no-info placeholder).
+		if uri == "" && name == "" {
+			continue
+		}
+		att := map[string]interface{}{"kind": kind}
+		if uri != "" {
+			att["uri"] = uri
+		}
+		if mimeType != "" {
+			att["mime_type"] = mimeType
+		}
+		if name != "" {
+			att["name"] = name
+		}
+		out = append(out, att)
+	}
+	if len(out) == 0 {
+		return nil
+	}
+	return out
+}
+
+// extractAttachmentFromFlatUploadManifest handles the canvas
+// chat_upload_receive shape: a single upload manifest at the root of
+// request_body with no JSON-RPC envelope. Canvas uses camelCase
+// `mimeType`; we normalize to snake_case `mime_type` on emit so the
+// wire shape matches the message-parts arm. Kind is derived from the
+// mime prefix (image/* → "image", audio/* → "audio", video/* → "video",
+// anything else → "file") because the canvas upload row doesn't carry
+// an explicit discriminator. Returns nil if neither `uri` nor `file_id`
+// is present at the root (i.e. not a flat upload manifest).
+func extractAttachmentFromFlatUploadManifest(body map[string]interface{}) map[string]interface{} {
+	uri, _ := body["uri"].(string)
+	fileID, _ := body["file_id"].(string)
+	if uri == "" && fileID == "" {
+		return nil
+	}
+	mimeType, _ := body["mimeType"].(string)
+	if mimeType == "" {
+		// Defensive: future canvas versions might emit snake_case directly.
+		mimeType, _ = body["mime_type"].(string)
+	}
+	name, _ := body["name"].(string)
+	// Apply the same minimum-info rule as the message-parts arm: a
+	// manifest with neither uri nor name is non-actionable; skip.
+	if uri == "" && name == "" {
+		return nil
+	}
+	att := map[string]interface{}{"kind": kindFromMimeType(mimeType)}
+	if uri != "" {
+		att["uri"] = uri
+	}
+	if mimeType != "" {
+		att["mime_type"] = mimeType
+	}
+	if name != "" {
+		att["name"] = name
+	}
+	return att
+}
+
+// kindFromMimeType derives the attachment `kind` discriminator from a
+// MIME type. Used by the flat-upload-manifest arm where the source row
+// has no explicit kind field.
+func kindFromMimeType(mime string) string {
+	switch {
+	case strings.HasPrefix(mime, "image/"):
+		return "image"
+	case strings.HasPrefix(mime, "audio/"):
+		return "audio"
+	case strings.HasPrefix(mime, "video/"):
+		return "video"
+	default:
+		return "file"
+	}
+}
+
+// includeFlagSet returns true iff `flag` appears in the comma-separated
+// `?include=` query value. Whitespace around entries is tolerated.
+// Empty `include` returns false (existing back-compat shape).
+//
+// The comma-separable form lets future fields ("attachments_only",
+// "tool_trace_expanded", etc.) slot in without further URL-param creep.
+func includeFlagSet(includeQuery, flag string) bool {
+	if includeQuery == "" || flag == "" {
+		return false
+	}
+	for _, raw := range strings.Split(includeQuery, ",") {
+		if strings.TrimSpace(raw) == flag {
+			return true
+		}
+	}
+	return false
+}
+
+// List handles GET /workspaces/:id/activity?type=&source=&limit=&since_secs=&since_id=&include=
+//
+// The `include` query param is comma-separable; today the only flag is
+// `peer_info`, which enriches a2a_receive rows with `peer_name`,
+// `peer_role`, `agent_card_url`, and an `attachments[]` projection (see
+// extractAttachmentsFromRequestBody). It's additive + opt-in — existing
+// callers that don't pass `?include=peer_info` see the unchanged shape.
+// Surface for the layered enrichment that lets Claude Code channel
+// pushes carry full sender identity instead of bare UUIDs (sibling
+// repos: molecule-ai-workspace-runtime + molecule-mcp-claude-channel).
 //
 // since_secs filters to activity_logs.created_at >= NOW() - INTERVAL '$N seconds'.
 // Optional, additive — callers that don't pass it get today's behavior (the
@@ -102,6 +308,8 @@ func (h *ActivityHandler) List(c *gin.Context) {
 	sinceSecsStr := c.Query("since_secs")
 	sinceID := c.Query("since_id")
 	beforeTSStr := c.Query("before_ts") // optional RFC3339 — return rows strictly older than this timestamp
+	include := c.Query("include")       // comma-separated; today's only flag is "peer_info"
+	includePeerInfo := includeFlagSet(include, "peer_info")

 	// Validate peer_id as a UUID at the trust boundary so a malformed
 	// caller (the agent or a downstream MCP tool) can't smuggle SQL
@@ -192,22 +400,60 @@ func (h *ActivityHandler) List(c *gin.Context) {
 		usingCursor = true
 	}

-	// Build query with optional filters
-	query := `SELECT id, workspace_id, activity_type, source_id, target_id, method,
-			   summary, request_body, response_body, tool_trace, duration_ms, status, error_detail, created_at
-		FROM activity_logs WHERE workspace_id = $1`
+	// Build query with optional filters. When ?include=peer_info is set,
+	// LEFT JOIN workspaces ON activity_logs.source_id = w.id so we can
+	// surface w.name + w.role on the row. LEFT (not INNER) is required
+	// for two reasons:
+	//   1. Canvas rows have source_id IS NULL — those must still appear
+	//      in the result set (with NULL peer_name/peer_role).
+	//   2. A peer workspace may have been deleted since the row was
+	//      written (no FK constraint on activity_logs.source_id) —
+	//      LEFT JOIN preserves the activity row with NULL peer fields
+	//      rather than silently dropping the row.
+	//
+	// agent_card_url is NOT pulled from the workspaces table; it's
+	// computed server-side from externalPlatformURL + source_id at
+	// projection time (mirrors molecule-ai-workspace-runtime
+	// a2a_client._agent_card_url_for which constructs
+	// {PLATFORM_URL}/registry/discover/{peer_id}).
+	//
+	// Column qualification (`activity_logs.<col>`) is added ONLY when
+	// the JOIN is present — disambiguates `id` / `created_at` which
+	// exist in both tables. When the JOIN is absent, unqualified
+	// column references preserve the exact wire-shape existing callers
+	// + existing test fixtures expect (back-compat).
+	actCol := ""
+	if includePeerInfo {
+		actCol = "activity_logs."
+	}
+	selectClause := `SELECT ` + actCol + `id, ` + actCol + `workspace_id, ` + actCol + `activity_type, ` +
+		actCol + `source_id, ` + actCol + `target_id, ` + actCol + `method, ` +
+		actCol + `summary, ` + actCol + `request_body, ` + actCol + `response_body, ` +
+		actCol + `tool_trace, ` + actCol + `duration_ms, ` + actCol + `status, ` +
+		actCol + `error_detail, ` + actCol + `created_at`
+	fromClause := ` FROM activity_logs`
+	if includePeerInfo {
+		selectClause += `, w.name AS peer_name, w.role AS peer_role`
+		fromClause += ` LEFT JOIN workspaces w ON w.id = activity_logs.source_id`
+	}
+	query := selectClause + fromClause + ` WHERE ` + actCol + `workspace_id = $1`
 	args := []interface{}{workspaceID}
 	argIdx := 2

+	// WHERE/ORDER column refs use the same `actCol` qualifier prefix
+	// computed above — empty string when no JOIN (back-compat with
+	// existing wire shape + sqlmock-regex test fixtures), or
+	// `activity_logs.` when LEFT JOIN'd (disambiguates `id` /
+	// `created_at` between the two tables).
 	if activityType != "" {
-		query += fmt.Sprintf(" AND activity_type = $%d", argIdx)
+		query += fmt.Sprintf(" AND "+actCol+"activity_type = $%d", argIdx)
 		args = append(args, activityType)
 		argIdx++
 	}
 	if source == "canvas" {
-		query += " AND source_id IS NULL"
+		query += " AND " + actCol + "source_id IS NULL"
 	} else if source == "agent" {
-		query += " AND source_id IS NOT NULL"
+		query += " AND " + actCol + "source_id IS NOT NULL"
 	} else if source != "" {
 		c.JSON(http.StatusBadRequest, gin.H{"error": "source must be 'canvas' or 'agent'"})
 		return
@@ -224,7 +470,7 @@ func (h *ActivityHandler) List(c *gin.Context) {
 		// and avoids duplicate parameter binding (some drivers reject the
 		// same arg slot reused, ours is fine but the explicit form is
 		// clearer to read and matches the rest of the builder.)
-		query += fmt.Sprintf(" AND (source_id = $%d OR target_id = $%d)", argIdx, argIdx)
+		query += fmt.Sprintf(" AND ("+actCol+"source_id = $%d OR "+actCol+"target_id = $%d)", argIdx, argIdx)
 		args = append(args, peerID)
 		argIdx++
 	}
@@ -232,7 +478,7 @@ func (h *ActivityHandler) List(c *gin.Context) {
 		// Strictly older — never replay a row with the exact same
 		// timestamp, mirrors the `created_at > cursorTime` shape
 		// `since_id` uses for forward paging.
-		query += fmt.Sprintf(" AND created_at < $%d", argIdx)
+		query += fmt.Sprintf(" AND "+actCol+"created_at < $%d", argIdx)
 		args = append(args, beforeTS)
 		argIdx++
 	}
@@ -241,13 +487,13 @@ func (h *ActivityHandler) List(c *gin.Context) {
 		// interpolated into the SQL string. `make_interval(secs => $N)`
 		// avoids the lib/pq quirk where INTERVAL '$N seconds' won't
 		// substitute a placeholder inside the literal.
-		query += fmt.Sprintf(" AND created_at >= NOW() - make_interval(secs => $%d)", argIdx)
+		query += fmt.Sprintf(" AND "+actCol+"created_at >= NOW() - make_interval(secs => $%d)", argIdx)
 		args = append(args, sinceSecs)
 		argIdx++
 	}
 	if usingCursor {
 		// Strictly after — never replay the cursor row itself.
-		query += fmt.Sprintf(" AND created_at > $%d", argIdx)
+		query += fmt.Sprintf(" AND "+actCol+"created_at > $%d", argIdx)
 		args = append(args, cursorTime)
 		argIdx++
 	}
@@ -257,9 +503,9 @@ func (h *ActivityHandler) List(c *gin.Context) {
 	// since_id) keeps DESC — that's the canvas/UI shape and changing it
 	// would surprise existing callers.
 	if usingCursor {
-		query += fmt.Sprintf(" ORDER BY created_at ASC LIMIT $%d", argIdx)
+		query += fmt.Sprintf(" ORDER BY "+actCol+"created_at ASC LIMIT $%d", argIdx)
 	} else {
-		query += fmt.Sprintf(" ORDER BY created_at DESC LIMIT $%d", argIdx)
+		query += fmt.Sprintf(" ORDER BY "+actCol+"created_at DESC LIMIT $%d", argIdx)
 	}
 	args = append(args, limit)

@@ -272,6 +518,14 @@ func (h *ActivityHandler) List(c *gin.Context) {
 	}
 	defer rows.Close()

+	// agent_card_url base computed once per request so we don't pay the
+	// header-read cost per row. Only meaningful when includePeerInfo is
+	// set; the empty string here is harmless when the flag is off.
+	var platformBase string
+	if includePeerInfo {
+		platformBase = externalPlatformURL(c)
+	}
+
 	activities := make([]map[string]interface{}, 0)
 	for rows.Next() {
 		var id, wsID, actType, status string
@@ -279,10 +533,23 @@ func (h *ActivityHandler) List(c *gin.Context) {
 		var reqBody, respBody, toolTrace []byte
 		var durationMs *int
 		var createdAt time.Time
+		// LEFT JOIN'd peer columns — pointer-string so a NULL row
+		// (canvas message OR deleted peer workspace) decodes as nil
+		// rather than empty-string. Only scanned when includePeerInfo
+		// is set (matched against the SELECT clause above).
+		var peerName, peerRole *string

-		if err := rows.Scan(&id, &wsID, &actType, &sourceID, &targetID, &method,
-			&summary, &reqBody, &respBody, &toolTrace, &durationMs, &status, &errorDetail, &createdAt); err != nil {
-			log.Printf("Activity scan error: %v", err)
+		var scanErr error
+		if includePeerInfo {
+			scanErr = rows.Scan(&id, &wsID, &actType, &sourceID, &targetID, &method,
+				&summary, &reqBody, &respBody, &toolTrace, &durationMs, &status, &errorDetail, &createdAt,
+				&peerName, &peerRole)
+		} else {
+			scanErr = rows.Scan(&id, &wsID, &actType, &sourceID, &targetID, &method,
+				&summary, &reqBody, &respBody, &toolTrace, &durationMs, &status, &errorDetail, &createdAt)
+		}
+		if scanErr != nil {
+			log.Printf("Activity scan error: %v", scanErr)
 			continue
 		}

@@ -308,6 +575,39 @@ func (h *ActivityHandler) List(c *gin.Context) {
 		if toolTrace != nil {
 			entry["tool_trace"] = json.RawMessage(toolTrace)
 		}
+
+		// peer_info enrichment (per ?include=peer_info). Only emit the
+		// new fields when the flag is set — back-compat for callers
+		// that don't request it.
+		if includePeerInfo {
+			// peer_name / peer_role: emit only when present (canvas
+			// rows have source_id IS NULL → peer_name is NULL by JOIN;
+			// also a peer workspace may have been deleted since the
+			// row was written → same NULL outcome). Omit-when-absent
+			// matches the Layer 3 adaptor's "spread when present"
+			// pattern; canvas_user rows legitimately have no peer_*.
+			if peerName != nil && *peerName != "" {
+				entry["peer_name"] = *peerName
+			}
+			if peerRole != nil && *peerRole != "" {
+				entry["peer_role"] = *peerRole
+			}
+			// agent_card_url: constructed server-side from
+			// externalPlatformURL + source_id. Mirrors the runtime-
+			// side helper a2a_client._agent_card_url_for which builds
+			// {PLATFORM_URL}/registry/discover/{peer_id}. Only set
+			// when source_id is present + non-empty.
+			if sourceID != nil && *sourceID != "" && platformBase != "" {
+				entry["agent_card_url"] = platformBase + "/registry/discover/" + *sourceID
+			}
+			// attachments: flatten file/image/audio parts from the
+			// request_body. nil when none — only project when
+			// non-empty so the omit-when-absent rule holds.
+			if atts := extractAttachmentsFromRequestBody(reqBody); len(atts) > 0 {
+				entry["attachments"] = atts
+			}
+		}
+
 		activities = append(activities, entry)
 	}
 	if err := rows.Err(); err != nil {
@@ -0,0 +1,701 @@
+package handlers
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/gin-gonic/gin"
+)
+
+// Tests for the `?include=peer_info` activity-feed enrichment.
+//
+// The enrichment is additive + opt-in. When the flag is absent, the
+// existing tests (TestActivityList_SourceCanvas, etc.) prove the wire
+// shape is unchanged. These tests prove:
+//   - When the flag IS set, the LEFT JOIN is issued and the SELECT
+//     adds w.name + w.role.
+//   - peer_name / peer_role surface from the joined row.
+//   - agent_card_url is composed server-side from
+//     externalPlatformURL + source_id and appears for non-canvas rows
+//     (source_id present).
+//   - attachments[] is projected from request_body.params.message.parts
+//     for file/image/audio parts.
+//   - Canvas rows (source_id NULL) do NOT get peer_name / peer_role /
+//     agent_card_url, but DO still appear in the result set (LEFT JOIN
+//     preserves them with NULL peer fields).
+//   - The `include` query param is comma-separable and only recognizes
+//     known flags.
+
+// ---------- includeFlagSet helper unit tests ----------
+
+func TestIncludeFlagSet(t *testing.T) {
+	cases := []struct {
+		query string
+		flag  string
+		want  bool
+	}{
+		{"", "peer_info", false},
+		{"peer_info", "peer_info", true},
+		{"peer_info,attachments", "peer_info", true},
+		{"attachments,peer_info", "peer_info", true},
+		{"attachments , peer_info ", "peer_info", true},
+		{"peer_infos", "peer_info", false},
+		{"peerinfo", "peer_info", false},
+		{"peer_info", "", false},
+		{",,", "peer_info", false},
+	}
+	for _, tc := range cases {
+		got := includeFlagSet(tc.query, tc.flag)
+		if got != tc.want {
+			t.Errorf("includeFlagSet(%q, %q) = %v, want %v", tc.query, tc.flag, got, tc.want)
+		}
+	}
+}
+
+// ---------- extractAttachmentsFromRequestBody unit tests ----------
+
+func TestExtractAttachmentsFromRequestBody_Empty(t *testing.T) {
+	if got := extractAttachmentsFromRequestBody(nil); got != nil {
+		t.Errorf("nil body: want nil, got %v", got)
+	}
+	if got := extractAttachmentsFromRequestBody([]byte("")); got != nil {
+		t.Errorf("empty body: want nil, got %v", got)
+	}
+	if got := extractAttachmentsFromRequestBody([]byte("not json")); got != nil {
+		t.Errorf("non-json body: want nil, got %v", got)
+	}
+}
+
+func TestExtractAttachmentsFromRequestBody_NoAttachments(t *testing.T) {
+	// Text-only message: no file/image/audio parts → nil
+	body := []byte(`{"jsonrpc":"2.0","method":"message/send","params":{"message":{"parts":[{"kind":"text","text":"hi"}]}}}`)
+	if got := extractAttachmentsFromRequestBody(body); got != nil {
+		t.Errorf("text-only: want nil, got %v", got)
+	}
+}
+
+func TestExtractAttachmentsFromRequestBody_FileKindV1(t *testing.T) {
+	// a2a-sdk v1 shape: kind=file, file:{uri,mime_type,name}
+	body := []byte(`{"jsonrpc":"2.0","method":"message/send","params":{"message":{"parts":[
+		{"kind":"text","text":"see attached"},
+		{"kind":"file","file":{"uri":"workspace:foo.pdf","mime_type":"application/pdf","name":"foo.pdf"}}
+	]}}}`)
+	atts := extractAttachmentsFromRequestBody(body)
+	if len(atts) != 1 {
+		t.Fatalf("want 1 attachment, got %d", len(atts))
+	}
+	if atts[0]["kind"] != "file" {
+		t.Errorf("kind: want file, got %v", atts[0]["kind"])
+	}
+	if atts[0]["uri"] != "workspace:foo.pdf" {
+		t.Errorf("uri mismatch: %v", atts[0]["uri"])
+	}
+	if atts[0]["mime_type"] != "application/pdf" {
+		t.Errorf("mime_type mismatch: %v", atts[0]["mime_type"])
+	}
+	if atts[0]["name"] != "foo.pdf" {
+		t.Errorf("name mismatch: %v", atts[0]["name"])
+	}
+}
+
+func TestExtractAttachmentsFromRequestBody_ImageAndAudio(t *testing.T) {
+	// Mixed image + audio parts; both surface
+	body := []byte(`{"jsonrpc":"2.0","method":"message/send","params":{"message":{"parts":[
+		{"kind":"image","file":{"uri":"workspace:a.png","mime_type":"image/png","name":"a.png"}},
+		{"kind":"audio","file":{"uri":"workspace:b.mp3","mime_type":"audio/mpeg","name":"b.mp3"}}
+	]}}}`)
+	atts := extractAttachmentsFromRequestBody(body)
+	if len(atts) != 2 {
+		t.Fatalf("want 2 attachments, got %d", len(atts))
+	}
+	if atts[0]["kind"] != "image" || atts[1]["kind"] != "audio" {
+		t.Errorf("kind order: got %v / %v", atts[0]["kind"], atts[1]["kind"])
+	}
+}
+
+func TestExtractAttachmentsFromRequestBody_LegacyV0TypeDiscriminator(t *testing.T) {
+	// Legacy v0 shape: type=file (not kind), inlined fields (no nested .file)
+	body := []byte(`{"jsonrpc":"2.0","method":"message/send","params":{"message":{"parts":[
+		{"type":"file","uri":"workspace:legacy.txt","mime_type":"text/plain","name":"legacy.txt"}
+	]}}}`)
+	atts := extractAttachmentsFromRequestBody(body)
+	if len(atts) != 1 {
+		t.Fatalf("want 1 attachment, got %d", len(atts))
+	}
+	if atts[0]["kind"] != "file" || atts[0]["uri"] != "workspace:legacy.txt" || atts[0]["name"] != "legacy.txt" {
+		t.Errorf("v0 part not surfaced: %v", atts[0])
+	}
+}
+
+func TestExtractAttachmentsFromRequestBody_SkipsEmptyParts(t *testing.T) {
+	// A "file" part with no uri AND no name is malformed — skip rather
+	// than emit a no-info entry.
+	body := []byte(`{"jsonrpc":"2.0","method":"message/send","params":{"message":{"parts":[
+		{"kind":"file","file":{}},
+		{"kind":"file","file":{"name":"only-name.bin"}}
+	]}}}`)
+	atts := extractAttachmentsFromRequestBody(body)
+	if len(atts) != 1 {
+		t.Fatalf("want 1 attachment (the named one), got %d", len(atts))
+	}
+	if atts[0]["name"] != "only-name.bin" {
+		t.Errorf("expected only-name.bin, got %v", atts[0])
+	}
+}
+
+func TestExtractAttachmentsFromRequestBody_MalformedShape(t *testing.T) {
+	// Various malformed shapes return nil (defensive)
+	for _, b := range []string{
+		`{}`,
+		`{"params":{}}`,
+		`{"params":{"message":{}}}`,
+		`{"params":{"message":{"parts":"not-a-list"}}}`,
+		`{"params":{"message":{"parts":[null,42,"string"]}}}`,
+	} {
+		if got := extractAttachmentsFromRequestBody([]byte(b)); got != nil {
+			t.Errorf("body %q: want nil, got %v", b, got)
+		}
+	}
+}
+
+// ---------- Activity List ?include=peer_info handler tests ----------
+
+func TestActivityList_IncludePeerInfo_IssuesLeftJoin(t *testing.T) {
+	// When ?include=peer_info is set, the query must:
+	//   1. SELECT include w.name + w.role aliased as peer_name/peer_role
+	//   2. FROM contains LEFT JOIN workspaces w ON w.id = activity_logs.source_id
+	//   3. WHERE uses qualified activity_logs.workspace_id (disambiguates
+	//      from workspaces.id post-JOIN)
+	//
+	// Pin all three so a future refactor can't silently drop the JOIN or
+	// the alias and have the test still pass.
+	mock := setupTestDB(t)
+	broadcaster := newTestBroadcaster()
+	handler := NewActivityHandler(broadcaster)
+
+	peerID := "11111111-2222-3333-4444-555555555555"
+	mock.ExpectQuery(
+		`SELECT .+w\.name AS peer_name, w\.role AS peer_role FROM activity_logs LEFT JOIN workspaces w ON w\.id = activity_logs\.source_id WHERE activity_logs\.workspace_id = .+`,
+	).
+		WithArgs("ws-1", 100).
+		WillReturnRows(sqlmock.NewRows([]string{
+			"id", "workspace_id", "activity_type", "source_id", "target_id",
+			"method", "summary", "request_body", "response_body",
+			"tool_trace", "duration_ms", "status", "error_detail", "created_at",
+			"peer_name", "peer_role",
+		}).
+			AddRow("act-1", "ws-1", "a2a_receive", peerID, "ws-1",
+				"message/send", "Agent message: hello",
+				[]byte(`{"jsonrpc":"2.0","method":"message/send","params":{"message":{"parts":[{"kind":"text","text":"hello"}]}}}`),
+				nil, nil, nil, "ok", nil, time.Now(),
+				"Production Manager", "product manager"))
+
+	gin.SetMode(gin.TestMode)
+	w := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(w)
+	c.Params = gin.Params{{Key: "id", Value: "ws-1"}}
+	c.Request = httptest.NewRequest("GET", "/workspaces/ws-1/activity?include=peer_info", nil)
+	c.Request.Host = "platform.test"
+	c.Request.Header.Set("X-Forwarded-Proto", "https")
+	handler.List(c)
+
+	if w.Code != http.StatusOK {
+		t.Fatalf("expected 200, 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("parse: %v", err)
+	}
+	if len(resp) != 1 {
+		t.Fatalf("want 1 row, got %d", len(resp))
+	}
+	r := resp[0]
+	if r["peer_name"] != "Production Manager" {
+		t.Errorf("peer_name: got %v", r["peer_name"])
+	}
+	if r["peer_role"] != "product manager" {
+		t.Errorf("peer_role: got %v", r["peer_role"])
+	}
+	wantURL := "https://platform.test/registry/discover/" + peerID
+	if r["agent_card_url"] != wantURL {
+		t.Errorf("agent_card_url: got %v, want %v", r["agent_card_url"], wantURL)
+	}
+	// Text-only message has no attachments → omit from envelope
+	if _, present := r["attachments"]; present {
+		t.Errorf("attachments should be omitted on text-only row; got %v", r["attachments"])
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Fatalf("unmet expectations: %v", err)
+	}
+}
+
+func TestActivityList_IncludePeerInfo_CanvasRowHasNoPeerFields(t *testing.T) {
+	// LEFT JOIN preserves canvas rows (source_id NULL) but their
+	// peer_name/peer_role come back as NULL — must omit from the
+	// envelope (not emit empty strings or null literals).
+	mock := setupTestDB(t)
+	broadcaster := newTestBroadcaster()
+	handler := NewActivityHandler(broadcaster)
+
+	mock.ExpectQuery(
+		`LEFT JOIN workspaces w ON w\.id = activity_logs\.source_id`,
+	).
+		WithArgs("ws-1", 100).
+		WillReturnRows(sqlmock.NewRows([]string{
+			"id", "workspace_id", "activity_type", "source_id", "target_id",
+			"method", "summary", "request_body", "response_body",
+			"tool_trace", "duration_ms", "status", "error_detail", "created_at",
+			"peer_name", "peer_role",
+		}).
+			// source_id NULL = canvas message; peer columns also NULL.
+			AddRow("act-canvas", "ws-1", "a2a_receive", nil, "ws-1",
+				"notify", "User said hi",
+				[]byte(`{"params":{"message":{"parts":[{"kind":"text","text":"hi"}]}}}`),
+				nil, nil, nil, "ok", nil, time.Now(),
+				nil, nil))
+
+	gin.SetMode(gin.TestMode)
+	w := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(w)
+	c.Params = gin.Params{{Key: "id", Value: "ws-1"}}
+	c.Request = httptest.NewRequest("GET", "/workspaces/ws-1/activity?include=peer_info", nil)
+	handler.List(c)
+
+	if w.Code != http.StatusOK {
+		t.Fatalf("expected 200, 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("parse: %v", err)
+	}
+	if len(resp) != 1 {
+		t.Fatalf("want 1 row, got %d", len(resp))
+	}
+	r := resp[0]
+	for _, k := range []string{"peer_name", "peer_role", "agent_card_url"} {
+		if _, present := r[k]; present {
+			t.Errorf("%s should be absent on canvas row; got %v", k, r[k])
+		}
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Fatalf("unmet expectations: %v", err)
+	}
+}
+
+func TestActivityList_IncludePeerInfo_AttachmentsSurfaceFromRequestBody(t *testing.T) {
+	// A peer_agent message with an inline file attachment must have
+	// attachments[] populated on the envelope.
+	mock := setupTestDB(t)
+	broadcaster := newTestBroadcaster()
+	handler := NewActivityHandler(broadcaster)
+
+	peerID := "11111111-2222-3333-4444-555555555555"
+	mock.ExpectQuery(`LEFT JOIN workspaces`).
+		WithArgs("ws-1", 100).
+		WillReturnRows(sqlmock.NewRows([]string{
+			"id", "workspace_id", "activity_type", "source_id", "target_id",
+			"method", "summary", "request_body", "response_body",
+			"tool_trace", "duration_ms", "status", "error_detail", "created_at",
+			"peer_name", "peer_role",
+		}).
+			AddRow("act-with-file", "ws-1", "a2a_receive", peerID, "ws-1",
+				"message/send", "Agent message: see attached",
+				[]byte(`{"jsonrpc":"2.0","method":"message/send","params":{"message":{"parts":[
+					{"kind":"text","text":"see attached"},
+					{"kind":"file","file":{"uri":"workspace:foo.pdf","mime_type":"application/pdf","name":"foo.pdf"}}
+				]}}}`),
+				nil, nil, nil, "ok", nil, time.Now(),
+				"Code Reviewer", "code reviewer"))
+
+	gin.SetMode(gin.TestMode)
+	w := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(w)
+	c.Params = gin.Params{{Key: "id", Value: "ws-1"}}
+	c.Request = httptest.NewRequest("GET", "/workspaces/ws-1/activity?include=peer_info", nil)
+	handler.List(c)
+
+	if w.Code != http.StatusOK {
+		t.Fatalf("expected 200, 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("parse: %v", err)
+	}
+	r := resp[0]
+	atts, ok := r["attachments"].([]interface{})
+	if !ok {
+		t.Fatalf("attachments missing or wrong type: %T %v", r["attachments"], r["attachments"])
+	}
+	if len(atts) != 1 {
+		t.Fatalf("want 1 attachment, got %d: %v", len(atts), atts)
+	}
+	att := atts[0].(map[string]interface{})
+	if att["kind"] != "file" || att["uri"] != "workspace:foo.pdf" || att["name"] != "foo.pdf" {
+		t.Errorf("attachment shape: %v", att)
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Fatalf("unmet expectations: %v", err)
+	}
+}
+
+func TestActivityList_IncludePeerInfo_Unset_NoJoinNoExtraFields(t *testing.T) {
+	// Back-compat — when ?include=peer_info is NOT passed, the SELECT
+	// uses unqualified column refs (no `activity_logs.` prefix) AND no
+	// JOIN. Existing tests pass this implicitly; this test pins it
+	// explicitly so a future refactor that accidentally turns the JOIN
+	// always-on gets caught.
+	mock := setupTestDB(t)
+	broadcaster := newTestBroadcaster()
+	handler := NewActivityHandler(broadcaster)
+
+	// Regex pinned: "FROM activity_logs WHERE workspace_id" — no JOIN
+	// keyword between FROM and WHERE; no `activity_logs.` qualifier on
+	// workspace_id.
+	mock.ExpectQuery(`SELECT id, workspace_id,.+ FROM activity_logs WHERE workspace_id = .+`).
+		WithArgs("ws-1", 100).
+		WillReturnRows(sqlmock.NewRows([]string{
+			"id", "workspace_id", "activity_type", "source_id", "target_id",
+			"method", "summary", "request_body", "response_body",
+			"tool_trace", "duration_ms", "status", "error_detail", "created_at",
+		}).
+			AddRow("act-1", "ws-1", "a2a_receive", "11111111-2222-3333-4444-555555555555", "ws-1",
+				"message/send", "Hello",
+				nil, nil, nil, nil, "ok", nil, time.Now()))
+
+	gin.SetMode(gin.TestMode)
+	w := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(w)
+	c.Params = gin.Params{{Key: "id", Value: "ws-1"}}
+	c.Request = httptest.NewRequest("GET", "/workspaces/ws-1/activity", nil)
+	handler.List(c)
+
+	if w.Code != http.StatusOK {
+		t.Fatalf("expected 200, 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("parse: %v", err)
+	}
+	if len(resp) != 1 {
+		t.Fatalf("want 1 row, got %d", len(resp))
+	}
+	// Confirm no peer_info enrichment leaks into the default envelope.
+	for _, k := range []string{"peer_name", "peer_role", "agent_card_url", "attachments"} {
+		if _, present := resp[0][k]; present {
+			t.Errorf("%s must NOT appear without ?include=peer_info; got %v", k, resp[0][k])
+		}
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Fatalf("unmet expectations: %v", err)
+	}
+}
+
+func TestActivityList_IncludePeerInfo_UnknownFlagIgnored(t *testing.T) {
+	// ?include=bogus must NOT issue the JOIN — only the recognized
+	// `peer_info` flag triggers enrichment. The unknown flag is silently
+	// ignored (additive, opt-in convention).
+	mock := setupTestDB(t)
+	broadcaster := newTestBroadcaster()
+	handler := NewActivityHandler(broadcaster)
+
+	mock.ExpectQuery(`SELECT id, workspace_id,.+ FROM activity_logs WHERE workspace_id = .+`).
+		WithArgs("ws-1", 100).
+		WillReturnRows(sqlmock.NewRows([]string{
+			"id", "workspace_id", "activity_type", "source_id", "target_id",
+			"method", "summary", "request_body", "response_body",
+			"tool_trace", "duration_ms", "status", "error_detail", "created_at",
+		}))
+
+	gin.SetMode(gin.TestMode)
+	w := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(w)
+	c.Params = gin.Params{{Key: "id", Value: "ws-1"}}
+	c.Request = httptest.NewRequest("GET", "/workspaces/ws-1/activity?include=bogus", nil)
+	handler.List(c)
+
+	if w.Code != http.StatusOK {
+		t.Fatalf("expected 200, got %d", w.Code)
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Fatalf("unmet expectations: %v", err)
+	}
+}
+
+// ---------- flat upload manifest (chat_upload_receive) tests ----------
+
+func TestKindFromMimeType(t *testing.T) {
+	cases := []struct {
+		mime string
+		want string
+	}{
+		{"image/png", "image"},
+		{"image/jpeg", "image"},
+		{"image/", "image"}, // prefix-only is still image
+		{"audio/mpeg", "audio"},
+		{"audio/wav", "audio"},
+		{"video/mp4", "video"},
+		{"video/webm", "video"},
+		{"application/pdf", "file"},
+		{"text/plain", "file"},
+		{"", "file"},
+		{"unknown", "file"},
+		{"image", "file"}, // no slash → not a prefix match
+	}
+	for _, tc := range cases {
+		if got := kindFromMimeType(tc.mime); got != tc.want {
+			t.Errorf("kindFromMimeType(%q) = %q, want %q", tc.mime, got, tc.want)
+		}
+	}
+}
+
+func TestExtractAttachmentsFromRequestBody_FlatUpload_Image(t *testing.T) {
+	// Canvas chat_upload_receive shape: flat manifest at request_body
+	// root with camelCase mimeType. The empirical example was a PNG
+	// pasted into the canvas; surfaces here with kind=image,
+	// mime_type=image/png (snake-case normalized), uri preserved.
+	body := []byte(`{
+		"uri":"platform-pending:091a9180-/26111d48-",
+		"name":"pasted-2026-05-21T23-12-25-0-0.png",
+		"size":677133,
+		"file_id":"26111d48-",
+		"mimeType":"image/png"
+	}`)
+	atts := extractAttachmentsFromRequestBody(body)
+	if len(atts) != 1 {
+		t.Fatalf("want 1 attachment, got %d: %v", len(atts), atts)
+	}
+	att := atts[0]
+	if att["kind"] != "image" {
+		t.Errorf("kind: want image, got %v", att["kind"])
+	}
+	if att["uri"] != "platform-pending:091a9180-/26111d48-" {
+		t.Errorf("uri: %v", att["uri"])
+	}
+	if att["mime_type"] != "image/png" {
+		t.Errorf("mime_type normalization (camelCase→snake_case) failed: %v", att["mime_type"])
+	}
+	if att["name"] != "pasted-2026-05-21T23-12-25-0-0.png" {
+		t.Errorf("name: %v", att["name"])
+	}
+	// camelCase `mimeType` MUST NOT leak into the projected envelope —
+	// only snake_case `mime_type` is the wire convention.
+	if _, present := att["mimeType"]; present {
+		t.Errorf("camelCase mimeType leaked into envelope: %v", att)
+	}
+	if _, present := att["file_id"]; present {
+		t.Errorf("file_id should not be surfaced on the attachment envelope (it's a canvas-internal id): %v", att)
+	}
+}
+
+func TestExtractAttachmentsFromRequestBody_FlatUpload_Audio(t *testing.T) {
+	body := []byte(`{"uri":"platform-pending:ws/file","name":"voice.mp3","file_id":"abc","mimeType":"audio/mpeg"}`)
+	atts := extractAttachmentsFromRequestBody(body)
+	if len(atts) != 1 || atts[0]["kind"] != "audio" {
+		t.Fatalf("want audio kind, got %v", atts)
+	}
+	if atts[0]["mime_type"] != "audio/mpeg" {
+		t.Errorf("mime_type: %v", atts[0]["mime_type"])
+	}
+}
+
+func TestExtractAttachmentsFromRequestBody_FlatUpload_Video(t *testing.T) {
+	body := []byte(`{"uri":"platform-pending:ws/file","name":"clip.mp4","file_id":"abc","mimeType":"video/mp4"}`)
+	atts := extractAttachmentsFromRequestBody(body)
+	if len(atts) != 1 || atts[0]["kind"] != "video" {
+		t.Fatalf("want video kind, got %v", atts)
+	}
+}
+
+func TestExtractAttachmentsFromRequestBody_FlatUpload_GenericFile(t *testing.T) {
+	// application/pdf has no image/audio/video prefix → kind=file
+	body := []byte(`{"uri":"platform-pending:ws/file","name":"doc.pdf","file_id":"abc","mimeType":"application/pdf"}`)
+	atts := extractAttachmentsFromRequestBody(body)
+	if len(atts) != 1 || atts[0]["kind"] != "file" {
+		t.Fatalf("want file kind, got %v", atts)
+	}
+}
+
+func TestExtractAttachmentsFromRequestBody_FlatUpload_NoMimeFallsToFile(t *testing.T) {
+	// No mimeType at all — kind defaults to "file", mime_type omitted.
+	body := []byte(`{"uri":"platform-pending:ws/file","name":"unknown.bin","file_id":"abc"}`)
+	atts := extractAttachmentsFromRequestBody(body)
+	if len(atts) != 1 {
+		t.Fatalf("want 1 attachment, got %d", len(atts))
+	}
+	if atts[0]["kind"] != "file" {
+		t.Errorf("kind: want file (default), got %v", atts[0]["kind"])
+	}
+	if _, present := atts[0]["mime_type"]; present {
+		t.Errorf("mime_type should be omitted when source has none, got %v", atts[0]["mime_type"])
+	}
+}
+
+func TestExtractAttachmentsFromRequestBody_FlatUpload_SnakeCaseMimeTypeAccepted(t *testing.T) {
+	// Defensive: a future canvas version (or non-canvas caller) that
+	// already emits snake_case mime_type should still be parsed.
+	body := []byte(`{"uri":"u","name":"n.png","mime_type":"image/png"}`)
+	atts := extractAttachmentsFromRequestBody(body)
+	if len(atts) != 1 {
+		t.Fatalf("want 1 attachment, got %d", len(atts))
+	}
+	if atts[0]["mime_type"] != "image/png" || atts[0]["kind"] != "image" {
+		t.Errorf("snake_case mime_type not honored: %v", atts[0])
+	}
+}
+
+func TestExtractAttachmentsFromRequestBody_FlatUpload_FileIDOnlyIsSkipped(t *testing.T) {
+	// file_id alone (no uri AND no name) is non-actionable — the
+	// downstream adaptor can't render a discoverable file from just an
+	// internal canvas id. Skip per the same minimum-info rule the
+	// message-parts arm applies to empty parts.
+	body := []byte(`{"file_id":"orphan-uuid","mimeType":"image/png"}`)
+	if got := extractAttachmentsFromRequestBody(body); got != nil {
+		t.Errorf("file_id-only manifest must be skipped, got %v", got)
+	}
+}
+
+func TestExtractAttachmentsFromRequestBody_FlatUpload_NameOnlyIsKept(t *testing.T) {
+	// Symmetric with the message-parts arm: a name without uri is still
+	// useful (the downstream adaptor can render "user uploaded foo.png").
+	body := []byte(`{"name":"only-name.bin","file_id":"abc","mimeType":"application/octet-stream"}`)
+	atts := extractAttachmentsFromRequestBody(body)
+	if len(atts) != 1 {
+		t.Fatalf("want 1 attachment, got %d", len(atts))
+	}
+	if atts[0]["name"] != "only-name.bin" {
+		t.Errorf("name not preserved: %v", atts[0])
+	}
+	if _, present := atts[0]["uri"]; present {
+		t.Errorf("uri should be omitted when absent in source, got %v", atts[0]["uri"])
+	}
+}
+
+func TestExtractAttachmentsFromRequestBody_MessagePartsTakesPrecedenceOverFlat(t *testing.T) {
+	// If a single request_body somehow has BOTH params.message.parts[]
+	// AND top-level uri/file_id (a pathological inbound), the
+	// message-parts arm wins — that's the documented inbound shape and
+	// it's been the only one historically extracted. The flat arm is a
+	// fallback for shapes that have NO parts.
+	body := []byte(`{
+		"uri":"platform-pending:should-not-win",
+		"file_id":"x",
+		"mimeType":"image/png",
+		"params":{"message":{"parts":[
+			{"kind":"file","file":{"uri":"workspace:should-win.pdf","mime_type":"application/pdf","name":"win.pdf"}}
+		]}}
+	}`)
+	atts := extractAttachmentsFromRequestBody(body)
+	if len(atts) != 1 {
+		t.Fatalf("want 1 attachment (from parts[]), got %d: %v", len(atts), atts)
+	}
+	if atts[0]["uri"] != "workspace:should-win.pdf" {
+		t.Errorf("message-parts arm did not take precedence: %v", atts[0])
+	}
+}
+
+func TestActivityList_IncludePeerInfo_ChatUploadReceiveCanvasRow(t *testing.T) {
+	// Wire-level integration: a canvas chat_upload_receive row (canvas
+	// user pasted an image) with source_id NULL (canvas message), flat
+	// upload manifest at request_body root. The `?include=peer_info`
+	// projection must surface attachments[] populated from the flat-
+	// upload-manifest arm while peer_name / peer_role / agent_card_url
+	// remain absent (canvas row has no peer).
+	mock := setupTestDB(t)
+	broadcaster := newTestBroadcaster()
+	handler := NewActivityHandler(broadcaster)
+
+	mock.ExpectQuery(`LEFT JOIN workspaces w ON w\.id = activity_logs\.source_id`).
+		WithArgs("ws-1", 100).
+		WillReturnRows(sqlmock.NewRows([]string{
+			"id", "workspace_id", "activity_type", "source_id", "target_id",
+			"method", "summary", "request_body", "response_body",
+			"tool_trace", "duration_ms", "status", "error_detail", "created_at",
+			"peer_name", "peer_role",
+		}).
+			// Empirical shape from 2026-05-21 ~23:12Z agents-team canvas paste.
+			AddRow("act-upload", "ws-1", "chat_upload_receive", nil, "ws-1",
+				"chat_upload_receive", "Canvas upload: pasted-2026-05-21T23-12-25-0-0.png",
+				[]byte(`{
+					"uri":"platform-pending:091a9180-b303-4a20-aefe-3a4a675b8aa4/26111d48-aaaa-bbbb-cccc-dddddddddddd",
+					"name":"pasted-2026-05-21T23-12-25-0-0.png",
+					"size":677133,
+					"file_id":"26111d48-aaaa-bbbb-cccc-dddddddddddd",
+					"mimeType":"image/png"
+				}`),
+				nil, nil, nil, "ok", nil, time.Now(),
+				nil, nil))
+
+	gin.SetMode(gin.TestMode)
+	w := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(w)
+	c.Params = gin.Params{{Key: "id", Value: "ws-1"}}
+	c.Request = httptest.NewRequest("GET", "/workspaces/ws-1/activity?include=peer_info", nil)
+	handler.List(c)
+
+	if w.Code != http.StatusOK {
+		t.Fatalf("expected 200, 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("parse: %v", err)
+	}
+	if len(resp) != 1 {
+		t.Fatalf("want 1 row, got %d", len(resp))
+	}
+	r := resp[0]
+	// Canvas row → no peer fields.
+	for _, k := range []string{"peer_name", "peer_role", "agent_card_url"} {
+		if _, present := r[k]; present {
+			t.Errorf("%s must NOT appear on canvas upload row; got %v", k, r[k])
+		}
+	}
+	// attachments[] populated from the flat-upload arm.
+	atts, ok := r["attachments"].([]interface{})
+	if !ok {
+		t.Fatalf("attachments missing or wrong type: %T %v", r["attachments"], r["attachments"])
+	}
+	if len(atts) != 1 {
+		t.Fatalf("want 1 attachment from flat manifest, got %d: %v", len(atts), atts)
+	}
+	att := atts[0].(map[string]interface{})
+	if att["kind"] != "image" {
+		t.Errorf("kind: want image (image/png prefix), got %v", att["kind"])
+	}
+	if att["mime_type"] != "image/png" {
+		t.Errorf("mime_type wire shape: want snake_case image/png, got %v", att["mime_type"])
+	}
+	if att["uri"] != "platform-pending:091a9180-b303-4a20-aefe-3a4a675b8aa4/26111d48-aaaa-bbbb-cccc-dddddddddddd" {
+		t.Errorf("uri preserved verbatim: got %v", att["uri"])
+	}
+	if att["name"] != "pasted-2026-05-21T23-12-25-0-0.png" {
+		t.Errorf("name: %v", att["name"])
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Fatalf("unmet expectations: %v", err)
+	}
+}
+
+// Sanity test using the existing test broadcaster setup — verifies the
+// extractAttachments helper round-trips through json.Marshal cleanly
+// (no map ordering issues, no type-coercion surprises).
+func TestExtractAttachmentsFromRequestBody_RoundTripsThroughJSON(t *testing.T) {
+	body := []byte(`{"params":{"message":{"parts":[{"kind":"file","file":{"uri":"workspace:r.bin","mime_type":"application/octet-stream","name":"r.bin"}}]}}}`)
+	atts := extractAttachmentsFromRequestBody(body)
+	b, err := json.Marshal(atts)
+	if err != nil {
+		t.Fatalf("marshal: %v", err)
+	}
+	var decoded []map[string]interface{}
+	if err := json.Unmarshal(b, &decoded); err != nil {
+		t.Fatalf("unmarshal: %v", err)
+	}
+	if len(decoded) != 1 || decoded[0]["uri"] != "workspace:r.bin" {
+		t.Fatalf("round-trip mismatch: %v", decoded)
+	}
+	_ = fmt.Sprintf // keep fmt import live if test trimming removes usage
+}