Compare commits
15 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 5599800bae | |||
| 494a1d8f9a | |||
| a356bc94f3 | |||
| 9981a5099a | |||
| 07d3dcd988 | |||
| 3ff613e3ad | |||
| 96c37cb098 | |||
| e123d07898 | |||
| 22fbf43580 | |||
| a47307969c | |||
| ff2557d899 | |||
| 119743d0de | |||
| c3806cd890 | |||
| 55e8c2d347 | |||
| 07b465f13d |
@@ -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 ))
|
||||
|
||||
+22
@@ -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
|
||||
}
|
||||
@@ -216,69 +216,102 @@ curl -fsS -X POST "{{PLATFORM_URL}}/registry/register" \
|
||||
const externalChannelTemplate = `# Claude Code channel — bridges this workspace's A2A traffic into your
|
||||
# Claude Code session. No tunnel/public URL needed (polling-based).
|
||||
#
|
||||
# Prereq: Bun installed (channel plugins are Bun scripts).
|
||||
# bun --version # must print a version number
|
||||
# Prereq: Bun 1.3+ installed (channel plugins are Bun scripts).
|
||||
# bun --version # must print a version (1.3.x or newer)
|
||||
#
|
||||
# 1. Inside Claude Code, install the channel plugin from its GitHub repo.
|
||||
# The plugin is NOT on Anthropic's default allowlist, so a one-time
|
||||
# marketplace-add is needed before install:
|
||||
# 1. Inside Claude Code, install the channel plugin. The plugin lives in
|
||||
# Molecule's own Gitea marketplace (not Anthropic's default), so a
|
||||
# one-time marketplace-add is needed before install:
|
||||
#
|
||||
# /plugin marketplace add https://git.moleculesai.app/molecule-ai/molecule-mcp-claude-channel.git
|
||||
# /plugin install molecule@molecule-channel
|
||||
#
|
||||
# Then either run /reload-plugins or restart Claude Code so the
|
||||
# plugin is registered.
|
||||
# Then /reload-plugins (or restart Claude Code) so the plugin is
|
||||
# registered.
|
||||
#
|
||||
# 2. Create the per-watched-workspace config file:
|
||||
# 2. Create (or extend) the per-host config file. The canonical SSOT
|
||||
# shape is MOLECULE_WORKSPACES_JSON — a JSON array of
|
||||
# {id, token, platform_url} objects. One plugin instance can watch
|
||||
# many workspaces across many tenants; append more objects to the
|
||||
# array (separate them with commas, NOT a newline):
|
||||
mkdir -p ~/.claude/channels/molecule
|
||||
cat > ~/.claude/channels/molecule/.env <<'EOF'
|
||||
MOLECULE_PLATFORM_URL={{PLATFORM_URL}}
|
||||
MOLECULE_WORKSPACE_IDS={{WORKSPACE_ID}}
|
||||
MOLECULE_WORKSPACE_TOKENS=<paste auth_token from create response>
|
||||
MOLECULE_WORKSPACES_JSON=[{"id":"{{WORKSPACE_ID}}","token":"<paste auth_token from create response>","platform_url":"{{PLATFORM_URL}}"}]
|
||||
EOF
|
||||
chmod 600 ~/.claude/channels/molecule/.env
|
||||
|
||||
# 3. Launch Claude Code with the channel enabled. Custom (non-Anthropic-
|
||||
# allowlisted) channels need the --dangerously-load-development-channels
|
||||
# flag to opt in — without it, you'll see "not on the approved channels
|
||||
# allowlist" on startup.
|
||||
claude --dangerously-load-development-channels \
|
||||
--channels plugin:molecule@molecule-channel
|
||||
# (Legacy single-platform shape — MOLECULE_PLATFORM_URL + comma-separated
|
||||
# MOLECULE_WORKSPACE_IDS + MOLECULE_WORKSPACE_TOKENS — is still supported
|
||||
# for back-compat but does NOT work across multiple tenant URLs. Use
|
||||
# MOLECULE_WORKSPACES_JSON above unless you have a specific reason.)
|
||||
|
||||
# 3. Launch Claude Code with the channel enabled. The channel spec is the
|
||||
# VALUE of --dangerously-load-development-channels — NOT a separate
|
||||
# --channels flag (that flag does not exist in current Claude Code;
|
||||
# passing it errors with "entries must be tagged: --channels").
|
||||
claude --dangerously-load-development-channels plugin:molecule@molecule-channel
|
||||
|
||||
# You should see on stderr:
|
||||
# molecule channel: connected — watching 1 workspace(s) at {{PLATFORM_URL}}
|
||||
# molecule channel: connected — watching N workspace(s) across M platform(s)
|
||||
# targets: <platform_url>: <workspace_id>
|
||||
#
|
||||
# Inbound A2A messages now surface as conversation turns. Claude's
|
||||
# replies route back via the reply_to_workspace MCP tool — no extra
|
||||
# wiring on your side.
|
||||
# Inbound A2A messages now surface as conversation turns (synthetic
|
||||
# <channel ...> tags). Claude's replies route back via the
|
||||
# reply_to_workspace / send_message_to_user MCP tools.
|
||||
#
|
||||
# Multi-workspace note: when watching more than one workspace, every
|
||||
# outbound tool call (send_message_to_user, reply_to_workspace,
|
||||
# delegate_task, list_peers) MUST pass _as_workspace=<id> so the plugin
|
||||
# knows which token to authenticate with. The host returns -32603 if you
|
||||
# forget — the synthetic <channel> tag's "watching_as" attribute tells
|
||||
# you which id to use.
|
||||
#
|
||||
# Common errors:
|
||||
# "plugin not installed" → Step 1 didn't run; run /plugin install
|
||||
# "plugin not installed" → Step 1 didn't run; run /plugin
|
||||
# marketplace add + /plugin install
|
||||
# inside Claude Code, then /reload-plugins.
|
||||
# "not on approved channels allowlist" → Add --dangerously-load-development-channels
|
||||
# to the launch command (Step 3).
|
||||
# "config-missing" → ~/.claude/channels/molecule/.env not
|
||||
# readable; re-run Step 2 and check chmod.
|
||||
# "entries must be tagged" → You passed --channels separately.
|
||||
# Put plugin:molecule@molecule-channel
|
||||
# directly after
|
||||
# --dangerously-load-development-channels.
|
||||
# "not on approved channels allowlist" → Org policy gating. See "managed
|
||||
# settings" note below.
|
||||
# "config-missing" → ~/.claude/channels/molecule/.env
|
||||
# not readable; re-run Step 2 and check
|
||||
# chmod 600.
|
||||
#
|
||||
# Team/Enterprise orgs: the --dangerously-load-development-channels flag is
|
||||
# blocked by managed settings. Your admin must set channelsEnabled=true and
|
||||
# add the plugin to allowedChannelPlugins in claude.ai admin settings.
|
||||
# Team/Enterprise plans: the channel allowlist is gated by org policy
|
||||
# AND must be written to the local managed-settings.json file on disk
|
||||
# (not the claude.ai web admin UI — there is no web toggle for this).
|
||||
# Path per OS:
|
||||
# macOS: /Library/Application Support/ClaudeCode/managed-settings.json
|
||||
# Linux: /etc/claude-code/managed-settings.json
|
||||
# Windows: C:\ProgramData\ClaudeCode\managed-settings.json
|
||||
# Set channelsEnabled: true and add
|
||||
# { "plugin": "molecule", "marketplace": "molecule-channel" }
|
||||
# to allowedChannelPlugins. Restart Claude Code after writing the file.
|
||||
# A user-level ~/.claude/settings.json does NOT work on Team/Enterprise
|
||||
# — this is the single most common reason a freshly-installed plugin
|
||||
# appears to do nothing.
|
||||
#
|
||||
# Multi-workspace: comma-separate IDs and tokens (same order). See
|
||||
# https://git.moleculesai.app/molecule-ai/molecule-mcp-claude-channel for
|
||||
# pairing flow, push-mode upgrade, and v0.2 roadmap.
|
||||
# Pro/Max plans skip the channelsEnabled gate but still need the
|
||||
# allowedChannelPlugins entry in the managed-settings file.
|
||||
|
||||
# Need help?
|
||||
# Documentation: https://doc.moleculesai.app/docs/guides/claude-code-channel-plugin
|
||||
# Full README: https://git.moleculesai.app/molecule-ai/molecule-mcp-claude-channel
|
||||
# Common errors:
|
||||
# • "plugin not installed" — run /plugin marketplace add then
|
||||
# /plugin install lines above; /reload-plugins or restart.
|
||||
# • "entries must be tagged: --channels" — the launch flag form
|
||||
# changed; use --dangerously-load-development-channels plugin:molecule@molecule-channel
|
||||
# (channel spec is the VALUE, not a separate --channels flag).
|
||||
# • "not on the approved channels allowlist" — custom channels need
|
||||
# --dangerously-load-development-channels; team/enterprise orgs
|
||||
# need admin to set channelsEnabled + allowedChannelPlugins.
|
||||
# allowedChannelPlugins in /Library/Application Support/ClaudeCode/managed-settings.json
|
||||
# (macOS) / equivalent on Linux+Windows. NOT a web setting.
|
||||
# • "Inbound messages not arriving" — stderr should show
|
||||
# "molecule channel: connected — watching N workspace(s)";
|
||||
# verify ~/.claude/channels/molecule/.env has PLATFORM_URL + token.
|
||||
# verify ~/.claude/channels/molecule/.env shape is MOLECULE_WORKSPACES_JSON.
|
||||
`
|
||||
|
||||
// externalUniversalMcpTemplate — runtime-agnostic standalone path.
|
||||
@@ -670,7 +703,15 @@ def heartbeat(client, url, ws, tok, start):
|
||||
r.raise_for_status()
|
||||
|
||||
def poll_inbound(client, url, ws, tok, since_id):
|
||||
params = {"since_secs": "30", "limit": "50"}
|
||||
# include=peer_info opts into Layer 1's row-level projection so each
|
||||
# polled activity carries peer_name, peer_role, agent_card_url, and
|
||||
# attachments[] inline (when source_id resolves to a peer / when the
|
||||
# message included a file). Pre-Layer-1 platforms ignore unknown query
|
||||
# params and return the bare row shape, so this is back-compat. Use
|
||||
# the extra fields in your reply logic — e.g. address the sender by
|
||||
# peer_name rather than UUID, or Read attached files via the workspace:
|
||||
# URIs in attachments[].
|
||||
params = {"since_secs": "30", "limit": "50", "include": "peer_info"}
|
||||
if since_id:
|
||||
params["since_id"] = since_id
|
||||
r = client.get(f"{url}/workspaces/{ws}/activity", params=params, headers=hdrs(url, tok))
|
||||
@@ -737,10 +778,16 @@ python3 ~/.molecule-ai/kimi-{{MCP_SERVER_NAME}}/kimi_bridge.py
|
||||
# What the script does:
|
||||
# • Registers the workspace in poll mode (no public URL needed)
|
||||
# • Heartbeats every 20s to keep STATUS = online on the canvas
|
||||
# • Polls /workspaces/:id/activity every 5s for new canvas messages
|
||||
# • Polls /workspaces/:id/activity?include=peer_info every 5s — Layer 1
|
||||
# enrichment surfaces peer_name / peer_role / agent_card_url /
|
||||
# attachments[] inline on each polled row when applicable
|
||||
# • Echo-replies via POST /workspaces/:id/notify
|
||||
#
|
||||
# To change the reply logic, edit the send_reply() call inside the loop.
|
||||
# Each polled item has top-level peer_name / peer_role / agent_card_url
|
||||
# fields (peer_agent rows) and attachments[] (any kind) when Layer 1 is
|
||||
# enabled on the platform — use them to disambiguate senders and to Read
|
||||
# attached files via the workspace: URIs.
|
||||
# To send a one-off reply from another terminal:
|
||||
# curl -fsS -X POST "{{PLATFORM_URL}}/workspaces/{{WORKSPACE_ID}}/notify" \
|
||||
# -H "Authorization: Bearer $(cat ~/.molecule-ai/kimi-{{MCP_SERVER_NAME}}/env | grep TOKEN | cut -d= -f2)" \
|
||||
|
||||
@@ -118,3 +118,86 @@ func TestExternalTemplates_NoBrokenMoleculeAIGitHubURLs(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestExternalChannelTemplate_LaunchFlagShape pins the Claude Code channel
|
||||
// snippet to the working launch invocation. The channel spec must be the
|
||||
// VALUE of --dangerously-load-development-channels, NOT a separate
|
||||
// --channels flag. The two-flag form (`--dangerously-load-development-channels
|
||||
// --channels plugin:molecule@...`) errors with "entries must be tagged:
|
||||
// --channels" on current Claude Code builds (2.1.143+) and silently no-ops
|
||||
// on older ones — either way, new users hit a wall on first launch.
|
||||
//
|
||||
// Empirical: hit by a session walking through this exact snippet 2026-05-21;
|
||||
// the broken form was copy-pasted from this template, ran, errored, and
|
||||
// confused the operator into believing the plugin install was broken when
|
||||
// the snippet itself was the bug.
|
||||
func TestExternalChannelTemplate_LaunchFlagShape(t *testing.T) {
|
||||
// The broken two-flag form. If this string ever appears in the
|
||||
// snippet again, the same onboarding pothole returns.
|
||||
bannedFormBroken := "--dangerously-load-development-channels \\\n --channels plugin:molecule@molecule-channel"
|
||||
if strings.Contains(externalChannelTemplate, bannedFormBroken) {
|
||||
t.Errorf("externalChannelTemplate contains the broken two-flag launch form. " +
|
||||
"Use --dangerously-load-development-channels plugin:molecule@molecule-channel (spec as value, not a separate --channels flag).")
|
||||
}
|
||||
|
||||
// The single-flag form must be present.
|
||||
requiredFormGood := "--dangerously-load-development-channels plugin:molecule@molecule-channel"
|
||||
if !strings.Contains(externalChannelTemplate, requiredFormGood) {
|
||||
t.Errorf("externalChannelTemplate must contain %q so operators see the working launch invocation", requiredFormGood)
|
||||
}
|
||||
}
|
||||
|
||||
// TestExternalChannelTemplate_CanonicalEnvShape pins the canvas-served
|
||||
// .env example to the canonical SSOT shape (MOLECULE_WORKSPACES_JSON)
|
||||
// rather than the legacy single-platform shape. The legacy form
|
||||
// (MOLECULE_PLATFORM_URL + comma-separated IDs/TOKENS) is still accepted
|
||||
// by the channel plugin's parseWorkspaceTargets but is single-tenant
|
||||
// only — it silently fails to onboard users who want to watch multiple
|
||||
// platforms (e.g. hongming + agents-team from the same plugin instance),
|
||||
// which is the post-PR#15 expected use case.
|
||||
func TestExternalChannelTemplate_CanonicalEnvShape(t *testing.T) {
|
||||
if !strings.Contains(externalChannelTemplate, "MOLECULE_WORKSPACES_JSON=") {
|
||||
t.Errorf("externalChannelTemplate must use MOLECULE_WORKSPACES_JSON as the canonical .env shape (the post-PR#15 SSOT)")
|
||||
}
|
||||
// The JSON example must contain the workspace_id + platform_url placeholders
|
||||
// so the canvas substitutes them at serve time.
|
||||
for _, ph := range []string{"{{WORKSPACE_ID}}", "{{PLATFORM_URL}}"} {
|
||||
if !strings.Contains(externalChannelTemplate, ph) {
|
||||
t.Errorf("externalChannelTemplate must contain placeholder %q so the canvas substitutes per-workspace values", ph)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestPollingTemplates_OptIntoPeerInfo pins the invariant that any template
|
||||
// which calls /workspaces/:id/activity for inbound delivery requests the
|
||||
// Layer 1 enrichment via ?include=peer_info. Without this opt-in, the
|
||||
// platform returns bare activity rows and the operator's bridge / channel
|
||||
// loses peer_name / peer_role / agent_card_url / attachments[] — they're
|
||||
// available on the server but not delivered.
|
||||
//
|
||||
// Pre-Layer-1 platforms ignore unknown query params (HTTP spec: filters
|
||||
// not understood are dropped), so this is back-compat across deploys.
|
||||
//
|
||||
// The Claude Code channel template doesn't include the poll URL in this
|
||||
// snippet — its polling lives in the plugin's own server.ts (handled by
|
||||
// molecule-mcp-claude-channel PR#21). The Kimi template DOES include a
|
||||
// poll loop in its kimi_bridge.py block, so the invariant applies there.
|
||||
func TestPollingTemplates_OptIntoPeerInfo(t *testing.T) {
|
||||
pollingTemplates := map[string]string{
|
||||
"externalKimiTemplate": externalKimiTemplate,
|
||||
}
|
||||
for name, body := range pollingTemplates {
|
||||
// If the snippet polls /activity, it must opt into peer_info.
|
||||
// The detection is intentionally loose ("/activity" appears in
|
||||
// the script) — operators who customize the script keep the
|
||||
// invariant only if the include hint is in the template.
|
||||
if !strings.Contains(body, "/activity") {
|
||||
t.Errorf("%s no longer polls /activity — review whether this test still applies", name)
|
||||
continue
|
||||
}
|
||||
if !strings.Contains(body, `"include": "peer_info"`) && !strings.Contains(body, "include=peer_info") {
|
||||
t.Errorf("%s polls /activity without ?include=peer_info — operators lose Layer 1 enrichment "+
|
||||
"(peer_name / peer_role / agent_card_url / attachments[]). Add the param to the poll URL.", name)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user