Compare commits

..

1 Commits

Author SHA1 Message Date
fullstack-engineer afc282c556 fix(channels): remove duplicate EncryptSensitiveFields + log json.Marshal errors
Runtime PR-Built Compatibility / PR-built wheel + import smoke (pull_request) Blocked by required conditions
Block internal-flavored paths / Block forbidden paths (pull_request) Successful in 28s
Harness Replays / detect-changes (pull_request) Successful in 31s
Secret scan / Scan diff for credential-shaped strings (pull_request) Successful in 34s
qa-review / approved (pull_request) Successful in 38s
security-review / approved (pull_request) Successful in 37s
CI / Detect changes (pull_request) Successful in 1m51s
lint-required-no-paths / lint-required-no-paths (pull_request) Successful in 1m41s
E2E API Smoke Test / detect-changes (pull_request) Successful in 1m56s
Handlers Postgres Integration / detect-changes (pull_request) Successful in 2m3s
Harness Replays / Harness Replays (pull_request) Successful in 15s
CI / Shellcheck (E2E scripts) (pull_request) Successful in 17s
CI / Python Lint & Test (pull_request) Successful in 15s
Runtime PR-Built Compatibility / detect-changes (pull_request) Failing after 10m50s
Handlers Postgres Integration / Handlers Postgres Integration (pull_request) Successful in 6m45s
E2E API Smoke Test / E2E API Smoke Test (pull_request) Failing after 10m58s
sop-tier-check / tier-check (pull_request) Successful in 41s
CI / Canvas (Next.js) (pull_request) Successful in 20m50s
CI / Platform (Go) (pull_request) Failing after 21m59s
CI / Canvas Deploy Reminder (pull_request) Has been skipped
gate-check-v3 / gate-check (pull_request) Failing after 14m21s
CI / all-required (pull_request) Failing after 10m32s
sop-checklist / all-items-acked (pull_request) [info tier:low] acked: 0/7 — missing: comprehensive-testing, local-postgres-e2e, staging-smoke, +4
audit-force-merge / audit (pull_request) Has been skipped
- Create: remove the duplicate EncryptSensitiveFields call introduced
  during OFFSEC-010 conflict resolution (commit 58882381). The
  function is idempotent so impact was nil (wasted CPU), but the block
  was dead code.
- Create: add error checks on json.Marshal for config and allowed_users
  (previously j, _ := json.Marshal(...) silently dropped marshal failures).
- Update: add error checks on json.Marshal for config and allowed_users
  (same silent-discard pattern). Also renamed the EncryptSensitiveFields
  error variable to encErr to avoid shadowing the outer err used by
  db.DB.ExecContext.

This PR complements #1109 (merge-queue: json.Unmarshal + rows.Err) and
#1110 (duplicate EncryptSensitiveFields removal) — those two cover the
same channels.go surface but neither adds json.Marshal error checks.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-15 04:18:09 +00:00
33 changed files with 209 additions and 1301 deletions
-26
View File
@@ -962,32 +962,6 @@ function MyChatPanel({ workspaceId, data }: Props) {
</div>
</div>
)}
{/* talk_to_user disabled banner — shown when the workspace has
talk_to_user_enabled=false. The agent cannot send canvas messages;
the user can re-enable the ability from here without opening settings. */}
{data.talkToUserEnabled === false && (
<div className="flex items-center gap-2 px-3 py-2 bg-surface-sunken border-b border-line/40 shrink-0">
<svg width="14" height="14" viewBox="0 0 16 16" fill="none" aria-hidden="true" className="shrink-0 text-ink-mid">
<path d="M8 1a7 7 0 1 0 0 14A7 7 0 0 0 8 1Zm0 10.5a.75.75 0 1 1 0-1.5.75.75 0 0 1 0 1.5ZM8 4a.75.75 0 0 1 .75.75v4a.75.75 0 0 1-1.5 0v-4A.75.75 0 0 1 8 4Z" fill="currentColor"/>
</svg>
<span className="text-[10px] text-ink-mid flex-1">
Agent is not enabled to chat with you.
</span>
<button
onClick={async () => {
try {
await api.patch(`/workspaces/${workspaceId}/abilities`, { talk_to_user_enabled: true });
useCanvasStore.getState().updateNodeData(workspaceId, { talkToUserEnabled: true });
} catch {
// ignore — user will see no change and can retry
}
}}
className="px-2 py-0.5 text-[10px] font-medium bg-accent/10 hover:bg-accent/20 text-accent rounded border border-accent/30 transition-colors shrink-0"
>
Enable
</button>
</div>
)}
{/* Messages */}
<div ref={containerRef} className="flex-1 overflow-y-auto p-3 space-y-3">
{loading && (
-4
View File
@@ -519,10 +519,6 @@ export function buildNodesAndEdges(
// #2054 — server-declared per-workspace provisioning timeout.
// Falls through to the runtime profile when null/absent.
provisionTimeoutMs: ws.provision_timeout_ms ?? null,
// Workspace abilities — defaults preserved for old platform versions
// that don't yet include these columns in the GET response.
broadcastEnabled: ws.broadcast_enabled ?? false,
talkToUserEnabled: ws.talk_to_user_enabled ?? true,
},
};
if (hasParent) {
-7
View File
@@ -99,13 +99,6 @@ export interface WorkspaceNodeData extends Record<string, unknown> {
* @/lib/runtimeProfiles. Lets a slow runtime declare its cold-boot
* expectation without a canvas release. */
provisionTimeoutMs?: number | null;
/** When true the workspace may POST /broadcast to send org-wide messages.
* Default false. Toggled by user/admin via PATCH /workspaces/:id/abilities. */
broadcastEnabled?: boolean;
/** When false the workspace cannot deliver canvas chat messages.
* send_message_to_user / POST /notify return 403 and the canvas
* shows a "not enabled" state with a button to re-enable. Default true. */
talkToUserEnabled?: boolean;
}
export type PanelTab = "details" | "skills" | "chat" | "terminal" | "config" | "schedule" | "channels" | "files" | "memory" | "traces" | "events" | "activity" | "audit";
-3
View File
@@ -299,9 +299,6 @@ export interface WorkspaceData {
* `@/lib/runtimeProfiles` when absent (the default behavior for any
* template that hasn't yet declared the field). */
provision_timeout_ms?: number | null;
/** Workspace ability flags (migration 20260514). */
broadcast_enabled?: boolean;
talk_to_user_enabled?: boolean;
}
let socket: ReconnectingSocket | null = null;
-296
View File
@@ -1,296 +0,0 @@
#!/usr/bin/env bash
# E2E test: workspace broadcast and talk-to-user platform abilities.
#
# What this proves:
# 1. talk_to_user_enabled (default true) — POST /notify works out-of-the-box.
# 2. PATCH /workspaces/:id/abilities { talk_to_user_enabled: false } disables
# delivery: /notify → 403 with error="talk_to_user_disabled" + delegate hint.
# 3. Re-enabling talk_to_user_enabled restores delivery.
# 4. broadcast_enabled (default false) — POST /broadcast → 403 when disabled.
# 5. PATCH { broadcast_enabled: true } enables fan-out.
# 6. POST /broadcast delivers to all non-sender, non-removed workspaces:
# - Returns {"status":"sent","delivered":N}
# - Receiver's activity log has a broadcast_receive entry with the message.
# - Sender's activity log has a broadcast_sent entry.
# 7. The sender itself does NOT receive a broadcast_receive entry.
#
# Usage: tests/e2e/test_workspace_abilities_e2e.sh
# Prereqs: workspace-server on http://localhost:8080, MOLECULE_ENV != production
set -euo pipefail
source "$(dirname "$0")/_lib.sh"
PASS=0
FAIL=0
SENDER_ID=""
RECEIVER_ID=""
cleanup() {
for wid in "$SENDER_ID" "$RECEIVER_ID"; do
if [ -n "$wid" ]; then
curl -s -X DELETE "$BASE/workspaces/$wid?confirm=true" > /dev/null || true
fi
done
}
trap cleanup EXIT INT TERM
assert() {
local label="$1" actual="$2" expected="$3"
if [ "$actual" = "$expected" ]; then
echo " PASS — $label"
PASS=$((PASS+1))
else
echo " FAIL — $label"
echo " expected: $expected"
echo " actual: $actual"
FAIL=$((FAIL+1))
fi
}
assert_contains() {
local label="$1" haystack="$2" needle="$3"
if echo "$haystack" | grep -qF "$needle"; then
echo " PASS — $label"
PASS=$((PASS+1))
else
echo " FAIL — $label"
echo " needle: $needle"
echo " haystack: $haystack"
FAIL=$((FAIL+1))
fi
}
assert_not_contains() {
local label="$1" haystack="$2" needle="$3"
if ! echo "$haystack" | grep -qF "$needle"; then
echo " PASS — $label"
PASS=$((PASS+1))
else
echo " FAIL — $label (unexpected match)"
echo " needle: $needle"
echo " haystack: $haystack"
FAIL=$((FAIL+1))
fi
}
# ── Pre-sweep: remove any stale leftover workspaces from a prior aborted run ──
echo "=== Setup ==="
for NAME in "Abilities Sender" "Abilities Receiver"; do
PRIOR=$(curl -s "$BASE/workspaces" | python3 -c "
import json, sys
try:
print(' '.join(w['id'] for w in json.load(sys.stdin) if w.get('name') == '$NAME'))
except Exception:
pass
")
for _wid in $PRIOR; do
echo "Sweeping leftover '$NAME' workspace: $_wid"
curl -s -X DELETE "$BASE/workspaces/$_wid?confirm=true" > /dev/null || true
done
done
R=$(curl -s -X POST "$BASE/workspaces" -H "Content-Type: application/json" \
-d '{"name":"Abilities Sender","tier":1}')
SENDER_ID=$(echo "$R" | python3 -c 'import json,sys;print(json.load(sys.stdin)["id"])' 2>/dev/null || true)
[ -n "$SENDER_ID" ] || { echo "Failed to create sender workspace: $R"; exit 1; }
echo "Created sender workspace: $SENDER_ID"
R=$(curl -s -X POST "$BASE/workspaces" -H "Content-Type: application/json" \
-d '{"name":"Abilities Receiver","tier":1}')
RECEIVER_ID=$(echo "$R" | python3 -c 'import json,sys;print(json.load(sys.stdin)["id"])' 2>/dev/null || true)
[ -n "$RECEIVER_ID" ] || { echo "Failed to create receiver workspace: $R"; exit 1; }
echo "Created receiver workspace: $RECEIVER_ID"
# Mint workspace-scoped bearer tokens (test-only endpoint, disabled in prod).
SENDER_TOKEN=$(e2e_mint_test_token "$SENDER_ID")
[ -n "$SENDER_TOKEN" ] || { echo "Failed to mint sender token"; exit 1; }
SENDER_AUTH="Authorization: Bearer $SENDER_TOKEN"
# Admin token — any live workspace bearer satisfies AdminAuth in local dev.
# In production-like envs, set MOLECULE_ADMIN_TOKEN.
ADMIN_TOKEN="${MOLECULE_ADMIN_TOKEN:-$SENDER_TOKEN}"
ADMIN_AUTH="Authorization: Bearer $ADMIN_TOKEN"
# ─────────────────────────────────────────────────────────────────────────────
echo ""
echo "=== Part 1: talk_to_user ability ==="
echo ""
echo "--- 1a: /notify works with default talk_to_user_enabled=true ---"
CODE=$(curl -s -o /dev/null -w "%{http_code}" -X POST "$BASE/workspaces/$SENDER_ID/notify" \
-H "Content-Type: application/json" -H "$SENDER_AUTH" \
-d '{"message":"Hello from sender"}')
assert "POST /notify returns 200 when talk_to_user_enabled=true (default)" "$CODE" "200"
echo ""
echo "--- 1b: Disable talk_to_user ---"
CODE=$(curl -s -o /dev/null -w "%{http_code}" -X PATCH "$BASE/workspaces/$SENDER_ID/abilities" \
-H "Content-Type: application/json" -H "$ADMIN_AUTH" \
-d '{"talk_to_user_enabled": false}')
assert "PATCH /abilities talk_to_user_enabled=false returns 200" "$CODE" "200"
# Verify the flag is reflected in the workspace GET response.
WS=$(curl -s "$BASE/workspaces/$SENDER_ID" -H "$SENDER_AUTH")
FLAG=$(echo "$WS" | python3 -c 'import json,sys;print(json.load(sys.stdin).get("talk_to_user_enabled","MISSING"))')
assert "GET /workspaces/:id reflects talk_to_user_enabled=false" "$FLAG" "False"
echo ""
echo "--- 1c: /notify blocked when talk_to_user disabled ---"
BODY=$(curl -s -w "" -X POST "$BASE/workspaces/$SENDER_ID/notify" \
-H "Content-Type: application/json" -H "$SENDER_AUTH" \
-d '{"message":"Should be blocked"}')
CODE=$(curl -s -o /dev/null -w "%{http_code}" -X POST "$BASE/workspaces/$SENDER_ID/notify" \
-H "Content-Type: application/json" -H "$SENDER_AUTH" \
-d '{"message":"Should be blocked"}')
assert "POST /notify returns 403 when talk_to_user_enabled=false" "$CODE" "403"
ERR=$(echo "$BODY" | python3 -c 'import json,sys;print(json.load(sys.stdin).get("error",""))' 2>/dev/null || echo "")
assert_contains "403 body contains talk_to_user_disabled error code" "$ERR" "talk_to_user_disabled"
HINT=$(echo "$BODY" | python3 -c 'import json,sys;print(json.load(sys.stdin).get("hint",""))' 2>/dev/null || echo "")
assert_contains "403 body contains delegate_task hint" "$HINT" "delegate_task"
echo ""
echo "--- 1d: Re-enable talk_to_user and verify /notify works again ---"
CODE=$(curl -s -o /dev/null -w "%{http_code}" -X PATCH "$BASE/workspaces/$SENDER_ID/abilities" \
-H "Content-Type: application/json" -H "$ADMIN_AUTH" \
-d '{"talk_to_user_enabled": true}')
assert "PATCH /abilities talk_to_user_enabled=true returns 200" "$CODE" "200"
CODE=$(curl -s -o /dev/null -w "%{http_code}" -X POST "$BASE/workspaces/$SENDER_ID/notify" \
-H "Content-Type: application/json" -H "$SENDER_AUTH" \
-d '{"message":"Re-enabled, should work"}')
assert "POST /notify returns 200 after re-enabling talk_to_user" "$CODE" "200"
# ─────────────────────────────────────────────────────────────────────────────
echo ""
echo "=== Part 2: broadcast ability ==="
echo ""
echo "--- 2a: Broadcast blocked by default (broadcast_enabled=false) ---"
CODE=$(curl -s -o /dev/null -w "%{http_code}" -X POST "$BASE/workspaces/$SENDER_ID/broadcast" \
-H "Content-Type: application/json" -H "$SENDER_AUTH" \
-d '{"message":"Should be blocked"}')
assert "POST /broadcast returns 403 when broadcast_enabled=false (default)" "$CODE" "403"
echo ""
echo "--- 2b: Enable broadcast ---"
CODE=$(curl -s -o /dev/null -w "%{http_code}" -X PATCH "$BASE/workspaces/$SENDER_ID/abilities" \
-H "Content-Type: application/json" -H "$ADMIN_AUTH" \
-d '{"broadcast_enabled": true}')
assert "PATCH /abilities broadcast_enabled=true returns 200" "$CODE" "200"
WS=$(curl -s "$BASE/workspaces/$SENDER_ID" -H "$SENDER_AUTH")
FLAG=$(echo "$WS" | python3 -c 'import json,sys;print(json.load(sys.stdin).get("broadcast_enabled","MISSING"))')
assert "GET /workspaces/:id reflects broadcast_enabled=true" "$FLAG" "True"
echo ""
echo "--- 2c: Successful broadcast fan-out ---"
BCAST=$(curl -s -X POST "$BASE/workspaces/$SENDER_ID/broadcast" \
-H "Content-Type: application/json" -H "$SENDER_AUTH" \
-d '{"message":"Org-wide notice: scheduled maintenance in 5 minutes."}')
BSTATUS=$(echo "$BCAST" | python3 -c 'import json,sys;print(json.load(sys.stdin).get("status",""))' 2>/dev/null || echo "")
BDELIVERED=$(echo "$BCAST" | python3 -c 'import json,sys;print(json.load(sys.stdin).get("delivered","-1"))' 2>/dev/null || echo "-1")
assert "POST /broadcast returns status=sent" "$BSTATUS" "sent"
# delivered count must be >= 1 (the receiver workspace).
echo " INFO — broadcast delivered=$BDELIVERED"
if python3 -c "import sys; sys.exit(0 if int('$BDELIVERED') >= 1 else 1)" 2>/dev/null; then
echo " PASS — delivered count >= 1"
PASS=$((PASS+1))
else
echo " FAIL — expected delivered >= 1, got $BDELIVERED"
FAIL=$((FAIL+1))
fi
echo ""
echo "--- 2d: Receiver activity log has broadcast_receive entry ---"
RECEIVER_TOKEN=$(e2e_mint_test_token "$RECEIVER_ID")
[ -n "$RECEIVER_TOKEN" ] || { echo "Failed to mint receiver token"; exit 1; }
RECEIVER_AUTH="Authorization: Bearer $RECEIVER_TOKEN"
ACT=$(curl -s -H "$RECEIVER_AUTH" "$BASE/workspaces/$RECEIVER_ID/activity?source=agent&limit=20")
ROW=$(echo "$ACT" | python3 -c '
import json, sys
rows = json.load(sys.stdin) or []
for r in rows:
if r.get("activity_type") == "broadcast_receive":
print(json.dumps(r))
break
')
[ -n "$ROW" ] || {
echo " FAIL — could not find broadcast_receive row in receiver activity"
FAIL=$((FAIL+1))
}
if [ -n "$ROW" ]; then
# Message is stored in summary field.
MSG=$(echo "$ROW" | python3 -c 'import json,sys;r=json.load(sys.stdin);print(r.get("summary",""))')
assert_contains "broadcast_receive row summary has original message" "$MSG" "scheduled maintenance"
# Sender ID is stored in source_id field.
SRC=$(echo "$ROW" | python3 -c 'import json,sys;r=json.load(sys.stdin);print(r.get("source_id",""))')
assert "broadcast_receive row source_id is sender workspace" "$SRC" "$SENDER_ID"
fi
echo ""
echo "--- 2e: Sender activity log has broadcast_sent entry ---"
ACT_SENDER=$(curl -s -H "$SENDER_AUTH" "$BASE/workspaces/$SENDER_ID/activity?limit=20")
SENT_ROW=$(echo "$ACT_SENDER" | python3 -c '
import json, sys
rows = json.load(sys.stdin) or []
for r in rows:
if r.get("activity_type") == "broadcast_sent":
print(json.dumps(r))
break
')
[ -n "$SENT_ROW" ] || {
echo " FAIL — could not find broadcast_sent row in sender activity"
FAIL=$((FAIL+1))
}
if [ -n "$SENT_ROW" ]; then
# Delivered count is baked into the summary field (no response_body for sender row).
SUMMARY=$(echo "$SENT_ROW" | python3 -c 'import json,sys;print(json.load(sys.stdin).get("summary",""))')
assert_contains "broadcast_sent summary mentions workspace count" "$SUMMARY" "workspace"
fi
echo ""
echo "--- 2f: Sender does NOT receive a broadcast_receive entry ---"
SELF_RECV=$(echo "$ACT_SENDER" | python3 -c '
import json, sys
rows = json.load(sys.stdin) or []
for r in rows:
if r.get("activity_type") == "broadcast_receive":
print("found")
break
')
assert_not_contains "sender has no broadcast_receive in own activity log" "${SELF_RECV:-}" "found"
# ─────────────────────────────────────────────────────────────────────────────
echo ""
echo "--- 2g: Empty message is rejected ---"
CODE=$(curl -s -o /dev/null -w "%{http_code}" -X POST "$BASE/workspaces/$SENDER_ID/broadcast" \
-H "Content-Type: application/json" -H "$SENDER_AUTH" \
-d '{"message":""}')
assert "POST /broadcast with empty message returns 400" "$CODE" "400"
echo ""
echo "--- 2h: Partial PATCH does not clobber other flags ---"
# Set talk_to_user=false, then patch only broadcast — talk_to_user must stay false.
curl -s -o /dev/null -X PATCH "$BASE/workspaces/$SENDER_ID/abilities" \
-H "Content-Type: application/json" -H "$ADMIN_AUTH" \
-d '{"talk_to_user_enabled": false}'
curl -s -o /dev/null -X PATCH "$BASE/workspaces/$SENDER_ID/abilities" \
-H "Content-Type: application/json" -H "$ADMIN_AUTH" \
-d '{"broadcast_enabled": false}'
WS=$(curl -s "$BASE/workspaces/$SENDER_ID" -H "$SENDER_AUTH")
TUF=$(echo "$WS" | python3 -c 'import json,sys;print(json.load(sys.stdin).get("talk_to_user_enabled","MISSING"))')
BEF=$(echo "$WS" | python3 -c 'import json,sys;print(json.load(sys.stdin).get("broadcast_enabled","MISSING"))')
assert "partial PATCH preserves talk_to_user_enabled=false" "$TUF" "False"
assert "partial PATCH sets broadcast_enabled=false" "$BEF" "False"
# ─────────────────────────────────────────────────────────────────────────────
echo ""
echo "=== Results: $PASS passed, $FAIL failed ==="
[ "$FAIL" -eq 0 ]
@@ -1,12 +1,7 @@
package handlers
import (
"context"
"database/sql"
"testing"
sqlmock "github.com/DATA-DOG/go-sqlmock"
"github.com/Molecule-AI/molecule-monorepo/platform/internal/db"
)
// TestExtractExpiresInSeconds covers the JSON parser used at enqueue time
@@ -63,207 +58,3 @@ func TestExtractExpiresInSeconds(t *testing.T) {
})
}
}
// ── QueueStatusByID ─────────────────────────────────────────────────────────────
func setupQueueStatusDB(t *testing.T) sqlmock.Sqlmock {
t.Helper()
mockDB, mock, err := sqlmock.New()
if err != nil {
t.Fatalf("failed to create sqlmock: %v", err)
}
prevDB := db.DB
db.DB = mockDB
t.Cleanup(func() { db.DB = prevDB; mockDB.Close() })
return mock
}
func TestQueueStatusByID_Success(t *testing.T) {
mock := setupQueueStatusDB(t)
queueID := "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
wsID := "cccccccc-cccc-cccc-cccc-cccccccccccc"
rows := sqlmock.NewRows([]string{
"id", "workspace_id", "status", "priority", "attempts",
"last_error", "enqueued_at", "dispatched_at", "completed_at", "expires_at",
"response_body",
}).AddRow(
queueID, wsID, "queued", 50, 0,
nil, // last_error
"2026-01-01T00:00:00Z", // enqueued_at
nil, // dispatched_at
nil, // completed_at
nil, // expires_at
nil, // response_body
)
mock.ExpectQuery(`SELECT`).
WithArgs(queueID).
WillReturnRows(rows)
qs, err := QueueStatusByID(context.Background(), queueID)
if err != nil {
t.Fatalf("QueueStatusByID returned error: %v", err)
}
if qs.ID != queueID {
t.Errorf("ID = %q, want %q", qs.ID, queueID)
}
if qs.WorkspaceID != wsID {
t.Errorf("WorkspaceID = %q, want %q", qs.WorkspaceID, wsID)
}
if qs.Status != "queued" {
t.Errorf("Status = %q, want %q", qs.Status, "queued")
}
if qs.Priority != 50 {
t.Errorf("Priority = %d, want 50", qs.Priority)
}
if qs.LastError != nil {
t.Errorf("LastError = %v, want nil", qs.LastError)
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("unmet sqlmock expectations: %v", err)
}
}
func TestQueueStatusByID_NotFound(t *testing.T) {
mock := setupQueueStatusDB(t)
queueID := "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
mock.ExpectQuery(`SELECT`).
WithArgs(queueID).
WillReturnError(sql.ErrNoRows)
qs, err := QueueStatusByID(context.Background(), queueID)
if err != sql.ErrNoRows {
t.Errorf("expected sql.ErrNoRows, got %v", err)
}
if qs != nil {
t.Errorf("expected nil queue status, got %+v", qs)
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("unmet sqlmock expectations: %v", err)
}
}
func TestQueueStatusByID_DBError(t *testing.T) {
mock := setupQueueStatusDB(t)
queueID := "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
mock.ExpectQuery(`SELECT`).
WithArgs(queueID).
WillReturnError(sql.ErrConnDone)
qs, err := QueueStatusByID(context.Background(), queueID)
if err == nil {
t.Error("expected error, got nil")
}
if qs != nil {
t.Errorf("expected nil queue status, got %+v", qs)
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("unmet sqlmock expectations: %v", err)
}
}
func TestQueueStatusByID_CompletedWithResponse(t *testing.T) {
mock := setupQueueStatusDB(t)
queueID := "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
wsID := "cccccccc-cccc-cccc-cccc-cccccccccccc"
respBody := []byte(`{"text":"delegation result"}`)
rows := sqlmock.NewRows([]string{
"id", "workspace_id", "status", "priority", "attempts",
"last_error", "enqueued_at", "dispatched_at", "completed_at", "expires_at",
"response_body",
}).AddRow(
queueID, wsID, "completed", 50, 1,
nil,
"2026-01-01T00:00:00Z",
"2026-01-01T00:01:00Z",
"2026-01-01T00:02:00Z",
nil,
respBody,
)
mock.ExpectQuery(`SELECT`).
WithArgs(queueID).
WillReturnRows(rows)
qs, err := QueueStatusByID(context.Background(), queueID)
if err != nil {
t.Fatalf("QueueStatusByID returned error: %v", err)
}
if qs.Status != "completed" {
t.Errorf("Status = %q, want completed", qs.Status)
}
if qs.ResponseBody == nil {
t.Fatal("ResponseBody should be set for completed status")
}
if string(qs.ResponseBody) != `{"text":"delegation result"}` {
t.Errorf("ResponseBody = %q, want %q", string(qs.ResponseBody), `{"text":"delegation result"}`)
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("unmet sqlmock expectations: %v", err)
}
}
// ── queueRowAuthFields ──────────────────────────────────────────────────────────
func TestQueueRowAuthFields_Success(t *testing.T) {
mock := setupQueueStatusDB(t)
queueID := "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
callerID := "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb"
wsID := "cccccccc-cccc-cccc-cccc-cccccccccccc"
rows := sqlmock.NewRows([]string{"caller_id", "workspace_id"}).
AddRow(callerID, wsID)
mock.ExpectQuery(`SELECT caller_id, workspace_id FROM a2a_queue WHERE id`).
WithArgs(queueID).
WillReturnRows(rows)
gotCaller, gotWs, err := queueRowAuthFields(context.Background(), queueID)
if err != nil {
t.Fatalf("queueRowAuthFields returned error: %v", err)
}
if gotCaller != callerID {
t.Errorf("callerID = %q, want %q", gotCaller, callerID)
}
if gotWs != wsID {
t.Errorf("workspaceID = %q, want %q", gotWs, wsID)
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("unmet sqlmock expectations: %v", err)
}
}
func TestQueueRowAuthFields_NotFound(t *testing.T) {
mock := setupQueueStatusDB(t)
queueID := "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
mock.ExpectQuery(`SELECT caller_id, workspace_id FROM a2a_queue WHERE id`).
WithArgs(queueID).
WillReturnError(sql.ErrNoRows)
_, _, err := queueRowAuthFields(context.Background(), queueID)
if err != sql.ErrNoRows {
t.Errorf("expected sql.ErrNoRows, got %v", err)
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("unmet sqlmock expectations: %v", err)
}
}
func TestQueueRowAuthFields_DBError(t *testing.T) {
mock := setupQueueStatusDB(t)
queueID := "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
mock.ExpectQuery(`SELECT caller_id, workspace_id FROM a2a_queue WHERE id`).
WithArgs(queueID).
WillReturnError(sql.ErrConnDone)
_, _, err := queueRowAuthFields(context.Background(), queueID)
if err == nil {
t.Error("expected error, got nil")
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("unmet sqlmock expectations: %v", err)
}
}
@@ -516,51 +516,3 @@ func TestDrainQueueForWorkspace_ClaimGuarding_SecondDrainGetsEmpty(t *testing.T)
t.Errorf("unmet sqlmock expectations: %v", err)
}
}
// ── QueueDepth ──────────────────────────────────────────────────────────────────
func TestQueueDepth_ReturnsCount(t *testing.T) {
mockDB, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherEqual))
if err != nil {
t.Fatalf("failed to create sqlmock: %v", err)
}
prevDB := db.DB
db.DB = mockDB
t.Cleanup(func() { db.DB = prevDB; mockDB.Close() })
wsID := "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
mock.ExpectQuery(`SELECT COUNT(*) FROM a2a_queue WHERE workspace_id = $1 AND status = 'queued'`).
WithArgs(wsID).
WillReturnRows(sqlmock.NewRows([]string{"count"}).AddRow(42))
got := QueueDepth(context.Background(), wsID)
if got != 42 {
t.Errorf("QueueDepth returned %d, want 42", got)
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("unmet sqlmock expectations: %v", err)
}
}
func TestQueueDepth_ZeroWhenEmpty(t *testing.T) {
mockDB, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherEqual))
if err != nil {
t.Fatalf("failed to create sqlmock: %v", err)
}
prevDB := db.DB
db.DB = mockDB
t.Cleanup(func() { db.DB = prevDB; mockDB.Close() })
wsID := "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
mock.ExpectQuery(`SELECT COUNT(*) FROM a2a_queue WHERE workspace_id = $1 AND status = 'queued'`).
WithArgs(wsID).
WillReturnRows(sqlmock.NewRows([]string{"count"}).AddRow(0))
got := QueueDepth(context.Background(), wsID)
if got != 0 {
t.Errorf("QueueDepth returned %d, want 0", got)
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("unmet sqlmock expectations: %v", err)
}
}
@@ -482,13 +482,6 @@ func (h *ActivityHandler) Notify(c *gin.Context) {
c.JSON(http.StatusNotFound, gin.H{"error": "workspace not found"})
return
}
if errors.Is(err, ErrTalkToUserDisabled) {
c.JSON(http.StatusForbidden, gin.H{
"error": "talk_to_user_disabled",
"hint": "This workspace is not allowed to send messages directly to the user. Forward your update to a parent workspace using delegate_task — they may be able to reach the user.",
})
return
}
c.JSON(http.StatusInternalServerError, gin.H{"error": "internal error"})
return
}
@@ -464,9 +464,9 @@ func TestNotify_PersistsToActivityLogsForReloadRecovery(t *testing.T) {
t.Cleanup(func() { db.DB = prevDB; mockDB.Close() })
// Workspace existence check
mock.ExpectQuery(`SELECT name, talk_to_user_enabled FROM workspaces`).
mock.ExpectQuery(`SELECT name FROM workspaces`).
WithArgs("ws-notify").
WillReturnRows(sqlmock.NewRows([]string{"name", "talk_to_user_enabled"}).AddRow("DD", true))
WillReturnRows(sqlmock.NewRows([]string{"name"}).AddRow("DD"))
// Persistence INSERT — verify shape
mock.ExpectExec(`INSERT INTO activity_logs`).
@@ -511,9 +511,9 @@ func TestNotify_WithAttachments_PersistsFilePartsForReload(t *testing.T) {
db.DB = mockDB
t.Cleanup(func() { db.DB = prevDB; mockDB.Close() })
mock.ExpectQuery(`SELECT name, talk_to_user_enabled FROM workspaces`).
mock.ExpectQuery(`SELECT name FROM workspaces`).
WithArgs("ws-attach").
WillReturnRows(sqlmock.NewRows([]string{"name", "talk_to_user_enabled"}).AddRow("DD", true))
WillReturnRows(sqlmock.NewRows([]string{"name"}).AddRow("DD"))
// Capture the JSONB arg so we can assert on the persisted shape
// AFTER the call (must include parts[].kind=file so reload
@@ -640,9 +640,9 @@ func TestNotify_DBFailure_StillBroadcastsAnd200(t *testing.T) {
db.DB = mockDB
t.Cleanup(func() { db.DB = prevDB; mockDB.Close() })
mock.ExpectQuery(`SELECT name, talk_to_user_enabled FROM workspaces`).
mock.ExpectQuery(`SELECT name FROM workspaces`).
WithArgs("ws-x").
WillReturnRows(sqlmock.NewRows([]string{"name", "talk_to_user_enabled"}).AddRow("DD", true))
WillReturnRows(sqlmock.NewRows([]string{"name"}).AddRow("DD"))
mock.ExpectExec(`INSERT INTO activity_logs`).
WillReturnError(fmt.Errorf("simulated db hiccup"))
@@ -54,11 +54,6 @@ import (
// timeout) surface as wrapped errors and should be treated as 503.
var ErrWorkspaceNotFound = errors.New("agent_message: workspace not found")
// ErrTalkToUserDisabled is returned when the workspace has
// talk_to_user_enabled=false. Callers surface HTTP 403 so the Python tool
// can detect it and suggest forwarding to a parent workspace.
var ErrTalkToUserDisabled = errors.New("agent_message: talk_to_user disabled")
// AgentMessageAttachment is one file attached to an agent → user
// message. Identical to handlers.NotifyAttachment in field set; kept
// distinct so the writer's API doesn't import a handler type with HTTP
@@ -112,20 +107,16 @@ func (w *AgentMessageWriter) Send(
// notify call surfaced as "workspace not found" and masked real
// incidents in the alert path.
var wsName string
var talkToUserEnabled bool
err := w.db.QueryRowContext(ctx,
`SELECT name, talk_to_user_enabled FROM workspaces WHERE id = $1 AND status != 'removed'`,
`SELECT name FROM workspaces WHERE id = $1 AND status != 'removed'`,
workspaceID,
).Scan(&wsName, &talkToUserEnabled)
).Scan(&wsName)
if errors.Is(err, sql.ErrNoRows) {
return ErrWorkspaceNotFound
}
if err != nil {
return fmt.Errorf("agent_message: workspace lookup: %w", err)
}
if !talkToUserEnabled {
return ErrTalkToUserDisabled
}
// 2. Build broadcast payload + WS-emit. Same shape that ChatTab's
// AGENT_MESSAGE handler in canvas/src/store/canvas-events.ts has
@@ -88,9 +88,9 @@ func TestAgentMessageWriter_Send_Success_NoAttachments(t *testing.T) {
mock := setupTestDB(t)
w := NewAgentMessageWriter(db.DB, newTestBroadcaster())
mock.ExpectQuery("SELECT name, talk_to_user_enabled FROM workspaces").
mock.ExpectQuery("SELECT name FROM workspaces").
WithArgs("ws-1").
WillReturnRows(sqlmock.NewRows([]string{"name", "talk_to_user_enabled"}).AddRow("CEO Ryan PC", true))
WillReturnRows(sqlmock.NewRows([]string{"name"}).AddRow("CEO Ryan PC"))
mock.ExpectExec(`INSERT INTO activity_logs.*'a2a_receive'.*'notify'`).
WithArgs(
@@ -116,9 +116,9 @@ func TestAgentMessageWriter_Send_Success_WithAttachments(t *testing.T) {
mock := setupTestDB(t)
w := NewAgentMessageWriter(db.DB, newTestBroadcaster())
mock.ExpectQuery("SELECT name, talk_to_user_enabled FROM workspaces").
mock.ExpectQuery("SELECT name FROM workspaces").
WithArgs("ws-att").
WillReturnRows(sqlmock.NewRows([]string{"name", "talk_to_user_enabled"}).AddRow("Ryan", true))
WillReturnRows(sqlmock.NewRows([]string{"name"}).AddRow("Ryan"))
mock.ExpectExec(`INSERT INTO activity_logs.*'a2a_receive'.*'notify'`).
WithArgs(
@@ -173,9 +173,9 @@ func TestAgentMessageWriter_Send_WorkspaceNotFound(t *testing.T) {
emitter := &capturingEmitter{}
w := NewAgentMessageWriter(db.DB, emitter)
mock.ExpectQuery("SELECT name, talk_to_user_enabled FROM workspaces").
mock.ExpectQuery("SELECT name FROM workspaces").
WithArgs("ws-missing").
WillReturnRows(sqlmock.NewRows([]string{"name", "talk_to_user_enabled"}))
WillReturnRows(sqlmock.NewRows([]string{"name"}))
err := w.Send(context.Background(), "ws-missing", "lost in the void", nil)
if !errors.Is(err, ErrWorkspaceNotFound) {
@@ -202,9 +202,9 @@ func TestAgentMessageWriter_Send_DBInsertFailureStillReturnsNil(t *testing.T) {
mock := setupTestDB(t)
w := NewAgentMessageWriter(db.DB, newTestBroadcaster())
mock.ExpectQuery("SELECT name, talk_to_user_enabled FROM workspaces").
mock.ExpectQuery("SELECT name FROM workspaces").
WithArgs("ws-dbfail").
WillReturnRows(sqlmock.NewRows([]string{"name", "talk_to_user_enabled"}).AddRow("CEO Ryan PC", true))
WillReturnRows(sqlmock.NewRows([]string{"name"}).AddRow("CEO Ryan PC"))
mock.ExpectExec(`INSERT INTO activity_logs`).
WillReturnError(errors.New("transient db error"))
@@ -223,9 +223,9 @@ func TestAgentMessageWriter_Send_PreviewTruncation(t *testing.T) {
mock := setupTestDB(t)
w := NewAgentMessageWriter(db.DB, newTestBroadcaster())
mock.ExpectQuery("SELECT name, talk_to_user_enabled FROM workspaces").
mock.ExpectQuery("SELECT name FROM workspaces").
WithArgs("ws-trunc").
WillReturnRows(sqlmock.NewRows([]string{"name", "talk_to_user_enabled"}).AddRow("Ryan", true))
WillReturnRows(sqlmock.NewRows([]string{"name"}).AddRow("Ryan"))
longMsg := strings.Repeat("x", 200)
mock.ExpectExec(`INSERT INTO activity_logs`).
@@ -263,9 +263,9 @@ func TestAgentMessageWriter_Send_BroadcastsAgentMessageEvent(t *testing.T) {
emitter := &capturingEmitter{}
w := NewAgentMessageWriter(db.DB, emitter)
mock.ExpectQuery("SELECT name, talk_to_user_enabled FROM workspaces").
mock.ExpectQuery("SELECT name FROM workspaces").
WithArgs("ws-bc").
WillReturnRows(sqlmock.NewRows([]string{"name", "talk_to_user_enabled"}).AddRow("Workspace Name", true))
WillReturnRows(sqlmock.NewRows([]string{"name"}).AddRow("Workspace Name"))
mock.ExpectExec(`INSERT INTO activity_logs`).
WillReturnResult(sqlmock.NewResult(1, 1))
@@ -315,7 +315,7 @@ func TestAgentMessageWriter_Send_DBErrorOnLookupReturnsWrapped(t *testing.T) {
w := NewAgentMessageWriter(db.DB, newTestBroadcaster())
transientErr := errors.New("connection refused")
mock.ExpectQuery("SELECT name, talk_to_user_enabled FROM workspaces").
mock.ExpectQuery("SELECT name FROM workspaces").
WithArgs("ws-dbdown").
WillReturnError(transientErr)
@@ -350,9 +350,9 @@ func TestAgentMessageWriter_Send_NonASCIIMessagePersists(t *testing.T) {
// the byte-slice bug.
msg := strings.Repeat("你", 200)
mock.ExpectQuery("SELECT name, talk_to_user_enabled FROM workspaces").
mock.ExpectQuery("SELECT name FROM workspaces").
WithArgs("ws-cjk").
WillReturnRows(sqlmock.NewRows([]string{"name", "talk_to_user_enabled"}).AddRow("CEO Ryan PC", true))
WillReturnRows(sqlmock.NewRows([]string{"name"}).AddRow("CEO Ryan PC"))
mock.ExpectExec(`INSERT INTO activity_logs`).
WithArgs(
@@ -395,9 +395,9 @@ func TestAgentMessageWriter_Send_OmitsAttachmentsKeyWhenEmpty(t *testing.T) {
emitter := &capturingEmitter{}
w := NewAgentMessageWriter(db.DB, emitter)
mock.ExpectQuery("SELECT name, talk_to_user_enabled FROM workspaces").
mock.ExpectQuery("SELECT name FROM workspaces").
WithArgs("ws-noatt").
WillReturnRows(sqlmock.NewRows([]string{"name", "talk_to_user_enabled"}).AddRow("X", true))
WillReturnRows(sqlmock.NewRows([]string{"name"}).AddRow("X"))
mock.ExpectExec(`INSERT INTO activity_logs`).
WillReturnResult(sqlmock.NewResult(1, 1))
+24 -13
View File
@@ -149,17 +149,18 @@ func (h *ChannelHandler) Create(c *gin.Context) {
return
}
// #319: encrypt sensitive fields (bot_token, webhook_secret) before
// persisting so a DB read/backup leak can't recover the credentials.
// Validation above ran against plaintext; storage is ciphertext.
if err := channels.EncryptSensitiveFields(body.Config); err != nil {
log.Printf("Channels: encrypt config failed for workspace %s: %v", workspaceID, err)
c.JSON(http.StatusInternalServerError, gin.H{"error": "encrypt failed"})
configJSON, mErr := json.Marshal(body.Config)
if mErr != nil {
log.Printf("Channels Create: marshal config for workspace %s: %v", workspaceID, mErr)
c.JSON(http.StatusInternalServerError, gin.H{"error": "marshal config failed"})
return
}
allowedJSON, mErr := json.Marshal(body.AllowedUsers)
if mErr != nil {
log.Printf("Channels Create: marshal allowed_users for workspace %s: %v", workspaceID, mErr)
c.JSON(http.StatusInternalServerError, gin.H{"error": "marshal allowed_users failed"})
return
}
configJSON, _ := json.Marshal(body.Config)
allowedJSON, _ := json.Marshal(body.AllowedUsers)
enabled := true
if body.Enabled != nil {
enabled = *body.Enabled
@@ -209,16 +210,26 @@ func (h *ChannelHandler) Update(c *gin.Context) {
// #319: re-encrypt sensitive fields on every config update — the
// PATCH body carries plaintext (client already had them plaintext in
// List response's unmasked path or typed fresh).
if err := channels.EncryptSensitiveFields(body.Config); err != nil {
log.Printf("Channels: encrypt update for workspace %s: %v", workspaceID, err)
if encErr := channels.EncryptSensitiveFields(body.Config); encErr != nil {
log.Printf("Channels: encrypt update for workspace %s: %v", workspaceID, encErr)
c.JSON(http.StatusInternalServerError, gin.H{"error": "encrypt failed"})
return
}
j, _ := json.Marshal(body.Config)
j, mErr := json.Marshal(body.Config)
if mErr != nil {
log.Printf("Channels Update: marshal config for channel %s: %v", channelID, mErr)
c.JSON(http.StatusInternalServerError, gin.H{"error": "marshal config failed"})
return
}
configArg = string(j)
}
if body.AllowedUsers != nil {
j, _ := json.Marshal(body.AllowedUsers)
j, mErr := json.Marshal(body.AllowedUsers)
if mErr != nil {
log.Printf("Channels Update: marshal allowed_users for channel %s: %v", channelID, mErr)
c.JSON(http.StatusInternalServerError, gin.H{"error": "marshal allowed_users failed"})
return
}
allowedArg = string(j)
}
@@ -947,73 +947,6 @@ func TestVerifyDiscordSignature_WrongLengthPubKey(t *testing.T) {
}
}
// ==================== matchesChatID pure function ====================
func TestMatchesChatID_ExactMatch(t *testing.T) {
cfg := map[string]interface{}{"chat_id": "123456"}
if !matchesChatID(cfg, "123456") {
t.Error("expected true for exact match")
}
}
func TestMatchesChatID_NoMatch(t *testing.T) {
cfg := map[string]interface{}{"chat_id": "123456"}
if matchesChatID(cfg, "654321") {
t.Error("expected false for non-matching chat ID")
}
}
func TestMatchesChatID_PrefixNoMatch(t *testing.T) {
// "123" is a prefix of "123456" but not an exact match.
cfg := map[string]interface{}{"chat_id": "123456"}
if matchesChatID(cfg, "123") {
t.Error("expected false for prefix of stored chat ID")
}
}
func TestMatchesChatID_CommaSeparatedMultiple(t *testing.T) {
cfg := map[string]interface{}{"chat_id": "111,222,333"}
for _, id := range []string{"111", "222", "333"} {
if !matchesChatID(cfg, id) {
t.Errorf("expected true for %q in comma-separated list", id)
}
}
if matchesChatID(cfg, "444") {
t.Error("expected false for ID not in list")
}
}
func TestMatchesChatID_WhitespaceTrimmed(t *testing.T) {
cfg := map[string]interface{}{"chat_id": "111, 222 , 333"}
if !matchesChatID(cfg, "222") {
t.Error("expected true for whitespace-trimmed match")
}
if matchesChatID(cfg, " 222") {
t.Error("expected false for whitespace in query (not trimmed from query)")
}
}
func TestMatchesChatID_EmptyChatID(t *testing.T) {
cfg := map[string]interface{}{"chat_id": ""}
if matchesChatID(cfg, "123456") {
t.Error("expected false for empty chat_id in config")
}
}
func TestMatchesChatID_MissingChatIDKey(t *testing.T) {
cfg := map[string]interface{}{}
if matchesChatID(cfg, "123456") {
t.Error("expected false when chat_id key is missing")
}
}
func TestMatchesChatID_NonStringChatID(t *testing.T) {
cfg := map[string]interface{}{"chat_id": 123456} // wrong type
if matchesChatID(cfg, "123456") {
t.Error("expected false when chat_id is not a string")
}
}
// TestChannelHandler_Webhook_Discord_NoKey_Returns401 verifies that a Discord
// webhook request is rejected with 401 when no public key is configured in the
// DB and DISCORD_APP_PUBLIC_KEY env var is not set.
@@ -230,21 +230,20 @@ func TestWorkspaceList_WithData(t *testing.T) {
broadcaster := newTestBroadcaster()
handler := NewWorkspaceHandler(broadcaster, nil, "http://localhost:8080", t.TempDir())
// 23 cols — broadcast_enabled + talk_to_user_enabled added after monthly_spend
// (migration 20260514). Column order must match scanWorkspaceRow exactly.
// 21 cols — see scanWorkspaceRow for order (max_concurrent_tasks
// lands between active_tasks and last_error_rate).
columns := []string{
"id", "name", "role", "tier", "status", "agent_card", "url",
"parent_id", "active_tasks", "max_concurrent_tasks",
"last_error_rate", "last_sample_error",
"uptime_seconds", "current_task", "runtime", "workspace_dir", "x", "y", "collapsed",
"budget_limit", "monthly_spend",
"broadcast_enabled", "talk_to_user_enabled",
}
rows := sqlmock.NewRows(columns).
AddRow("ws-1", "Agent One", "worker", 1, "online", []byte(`{"name":"agent1"}`), "http://localhost:8001",
nil, 3, 1, 0.02, "", 7200, "processing", "langgraph", "", 10.0, 20.0, false, nil, int64(0), false, true).
nil, 3, 1, 0.02, "", 7200, "processing", "langgraph", "", 10.0, 20.0, false, nil, int64(0)).
AddRow("ws-2", "Agent Two", "", 2, "degraded", []byte("null"), "",
nil, 0, 1, 0.6, "timeout", 100, "", "claude-code", "", 50.0, 60.0, true, nil, int64(0), false, true)
nil, 0, 1, 0.6, "timeout", 100, "", "claude-code", "", 50.0, 60.0, true, nil, int64(0))
mock.ExpectQuery("SELECT w.id, w.name").
WillReturnRows(rows)
@@ -392,21 +392,21 @@ func TestWorkspaceList(t *testing.T) {
broadcaster := newTestBroadcaster()
handler := NewWorkspaceHandler(broadcaster, nil, "http://localhost:8080", "/tmp/configs")
// 23 cols: broadcast_enabled + talk_to_user_enabled added after monthly_spend
// (migration 20260514). Column order must match scanWorkspaceRow exactly.
// 21 cols: `max_concurrent_tasks` added between active_tasks and
// last_error_rate (see scanWorkspaceRow + COALESCE(w.max_concurrent_tasks, 1)
// in workspace.go). Column order must match that scan exactly.
columns := []string{
"id", "name", "role", "tier", "status", "agent_card", "url",
"parent_id", "active_tasks", "max_concurrent_tasks",
"last_error_rate", "last_sample_error",
"uptime_seconds", "current_task", "runtime", "workspace_dir", "x", "y", "collapsed",
"budget_limit", "monthly_spend",
"broadcast_enabled", "talk_to_user_enabled",
}
rows := sqlmock.NewRows(columns).
AddRow("ws-1", "Agent One", "worker", 1, "online", []byte("null"), "http://localhost:8001",
nil, 0, 1, 0.0, "", 100, "", "claude-code", "", 10.0, 20.0, false, nil, int64(0), false, true).
nil, 0, 1, 0.0, "", 100, "", "claude-code", "", 10.0, 20.0, false, nil, int64(0)).
AddRow("ws-2", "Agent Two", "manager", 2, "provisioning", []byte("null"), "",
nil, 0, 1, 0.0, "", 0, "", "langgraph", "", 50.0, 60.0, false, nil, int64(0), false, true)
nil, 0, 1, 0.0, "", 0, "", "langgraph", "", 50.0, 60.0, false, nil, int64(0))
mock.ExpectQuery("SELECT w.id, w.name").
WillReturnRows(rows)
@@ -1120,14 +1120,13 @@ func TestWorkspaceGet_CurrentTask(t *testing.T) {
"parent_id", "active_tasks", "max_concurrent_tasks", "last_error_rate", "last_sample_error",
"uptime_seconds", "current_task", "runtime", "workspace_dir", "x", "y", "collapsed",
"budget_limit", "monthly_spend",
"broadcast_enabled", "talk_to_user_enabled",
}
mock.ExpectQuery("SELECT w.id, w.name").
WithArgs("dddddddd-0004-0000-0000-000000000000").
WillReturnRows(sqlmock.NewRows(columns).AddRow(
"dddddddd-0004-0000-0000-000000000000", "Task Worker", "worker", 1, "online", []byte("null"), "http://localhost:9000",
nil, 2, 1, 0.0, "", 300, "Analyzing document", "langgraph", "", 10.0, 20.0, false,
nil, int64(0), false, true,
nil, int64(0),
))
w := httptest.NewRecorder()
@@ -751,9 +751,9 @@ func TestMCPHandler_SendMessageToUser_DBErrorLogsAndStill200s(t *testing.T) {
t.Setenv("MOLECULE_MCP_ALLOW_SEND_MESSAGE", "true")
h, mock := newMCPHandler(t)
mock.ExpectQuery("SELECT name, talk_to_user_enabled FROM workspaces").
mock.ExpectQuery("SELECT name FROM workspaces").
WithArgs("ws-err").
WillReturnRows(sqlmock.NewRows([]string{"name", "talk_to_user_enabled"}).AddRow("CEO Ryan PC", true))
WillReturnRows(sqlmock.NewRows([]string{"name"}).AddRow("CEO Ryan PC"))
// INSERT fails — must NOT abort the tool response.
mock.ExpectExec(`INSERT INTO activity_logs.*'a2a_receive'.*'notify'`).
@@ -802,9 +802,9 @@ func TestMCPHandler_SendMessageToUser_ResponseBodyShape(t *testing.T) {
const userMessage = "Hi there from the agent"
mock.ExpectQuery("SELECT name, talk_to_user_enabled FROM workspaces").
mock.ExpectQuery("SELECT name FROM workspaces").
WithArgs("ws-shape").
WillReturnRows(sqlmock.NewRows([]string{"name", "talk_to_user_enabled"}).AddRow("CEO Ryan PC", true))
WillReturnRows(sqlmock.NewRows([]string{"name"}).AddRow("CEO Ryan PC"))
// Capture the response_body argument and assert its exact shape.
mock.ExpectExec(`INSERT INTO activity_logs.*'a2a_receive'.*'notify'`).
@@ -861,9 +861,9 @@ func TestMCPHandler_SendMessageToUser_PersistsToActivityLog(t *testing.T) {
// before it does anything else. Returning a name lets the
// broadcast payload populate; the test doesn't assert on the
// broadcast (no observable WS in this fake), only on the DB.
mock.ExpectQuery("SELECT name, talk_to_user_enabled FROM workspaces").
mock.ExpectQuery("SELECT name FROM workspaces").
WithArgs("ws-msg").
WillReturnRows(sqlmock.NewRows([]string{"name", "talk_to_user_enabled"}).AddRow("CEO Ryan PC", true))
WillReturnRows(sqlmock.NewRows([]string{"name"}).AddRow("CEO Ryan PC"))
// The persistence INSERT — pin the exact shape so a future
// refactor that switches columns or drops `method='notify'`
@@ -595,7 +595,7 @@ func scanWorkspaceRow(rows interface {
var id, name, role, status, url, sampleError, currentTask, runtime, workspaceDir string
var tier, activeTasks, maxConcurrentTasks, uptimeSeconds int
var errorRate, x, y float64
var collapsed, broadcastEnabled, talkToUserEnabled bool
var collapsed bool
var parentID *string
var agentCard []byte
var budgetLimit sql.NullInt64
@@ -604,7 +604,7 @@ func scanWorkspaceRow(rows interface {
err := rows.Scan(&id, &name, &role, &tier, &status, &agentCard, &url,
&parentID, &activeTasks, &maxConcurrentTasks, &errorRate, &sampleError, &uptimeSeconds,
&currentTask, &runtime, &workspaceDir, &x, &y, &collapsed,
&budgetLimit, &monthlySpend, &broadcastEnabled, &talkToUserEnabled)
&budgetLimit, &monthlySpend)
if err != nil {
return nil, err
}
@@ -628,8 +628,6 @@ func scanWorkspaceRow(rows interface {
"x": x,
"y": y,
"collapsed": collapsed,
"broadcast_enabled": broadcastEnabled,
"talk_to_user_enabled": talkToUserEnabled,
}
// budget_limit: nil when no limit set, int64 otherwise
@@ -665,8 +663,7 @@ const workspaceListQuery = `
COALESCE(w.current_task, ''), COALESCE(w.runtime, 'langgraph'),
COALESCE(w.workspace_dir, ''),
COALESCE(cl.x, 0), COALESCE(cl.y, 0), COALESCE(cl.collapsed, false),
w.budget_limit, COALESCE(w.monthly_spend, 0),
w.broadcast_enabled, w.talk_to_user_enabled
w.budget_limit, COALESCE(w.monthly_spend, 0)
FROM workspaces w
LEFT JOIN canvas_layouts cl ON cl.workspace_id = w.id
WHERE w.status != 'removed'
@@ -726,8 +723,7 @@ func (h *WorkspaceHandler) Get(c *gin.Context) {
COALESCE(w.current_task, ''), COALESCE(w.runtime, 'langgraph'),
COALESCE(w.workspace_dir, ''),
COALESCE(cl.x, 0), COALESCE(cl.y, 0), COALESCE(cl.collapsed, false),
w.budget_limit, COALESCE(w.monthly_spend, 0),
w.broadcast_enabled, w.talk_to_user_enabled
w.budget_limit, COALESCE(w.monthly_spend, 0)
FROM workspaces w
LEFT JOIN canvas_layouts cl ON cl.workspace_id = w.id
WHERE w.id = $1
@@ -1,82 +0,0 @@
package handlers
// workspace_abilities.go — PATCH /workspaces/:id/abilities
//
// Allows users and admin agents to toggle two workspace-level ability flags:
//
// broadcast_enabled — workspace may POST /broadcast to send org-wide messages
// talk_to_user_enabled — workspace may deliver canvas chat messages via
// send_message_to_user / POST /notify
//
// Gated behind AdminAuth so workspace agents cannot self-modify their own
// ability flags (that would let any agent grant itself broadcast rights or
// suppress its own chat-silence constraint).
import (
"log"
"net/http"
"github.com/Molecule-AI/molecule-monorepo/platform/internal/db"
"github.com/gin-gonic/gin"
)
// AbilitiesPayload carries the subset of ability flags the caller wants to
// update. Fields are pointers so that the handler can distinguish "caller
// supplied false" from "caller omitted the field" (omitempty semantics).
type AbilitiesPayload struct {
BroadcastEnabled *bool `json:"broadcast_enabled"`
TalkToUserEnabled *bool `json:"talk_to_user_enabled"`
}
// PatchAbilities handles PATCH /workspaces/:id/abilities (AdminAuth).
func PatchAbilities(c *gin.Context) {
id := c.Param("id")
if err := validateWorkspaceID(id); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "invalid workspace ID"})
return
}
var body AbilitiesPayload
if err := c.ShouldBindJSON(&body); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request body"})
return
}
if body.BroadcastEnabled == nil && body.TalkToUserEnabled == nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "at least one ability field required"})
return
}
ctx := c.Request.Context()
var exists bool
if err := db.DB.QueryRowContext(ctx,
`SELECT EXISTS(SELECT 1 FROM workspaces WHERE id = $1 AND status != 'removed')`, id,
).Scan(&exists); err != nil || !exists {
c.JSON(http.StatusNotFound, gin.H{"error": "workspace not found"})
return
}
if body.BroadcastEnabled != nil {
if _, err := db.DB.ExecContext(ctx,
`UPDATE workspaces SET broadcast_enabled = $2, updated_at = now() WHERE id = $1`,
id, *body.BroadcastEnabled,
); err != nil {
log.Printf("PatchAbilities broadcast_enabled for %s: %v", id, err)
c.JSON(http.StatusInternalServerError, gin.H{"error": "update failed"})
return
}
}
if body.TalkToUserEnabled != nil {
if _, err := db.DB.ExecContext(ctx,
`UPDATE workspaces SET talk_to_user_enabled = $2, updated_at = now() WHERE id = $1`,
id, *body.TalkToUserEnabled,
); err != nil {
log.Printf("PatchAbilities talk_to_user_enabled for %s: %v", id, err)
c.JSON(http.StatusInternalServerError, gin.H{"error": "update failed"})
return
}
}
c.JSON(http.StatusOK, gin.H{"status": "updated"})
}
@@ -1,142 +0,0 @@
package handlers
// workspace_broadcast.go — POST /workspaces/:id/broadcast
//
// Allows a workspace with broadcast_enabled=true to send a message to every
// non-removed agent workspace in the org. The message is:
//
// • Persisted in each recipient's activity_logs (type='broadcast_receive')
// so poll-mode agents pick it up via GET /activity.
// • Broadcast via WebSocket BROADCAST_MESSAGE event so canvas panels can
// show a real-time banner for each recipient workspace.
//
// The sender's own workspace logs a 'broadcast_sent' activity row for
// traceability.
//
// Auth: WorkspaceAuth (the agent triggers this with its own bearer token).
// The handler re-validates broadcast_enabled inside the DB lookup to prevent
// TOCTOU — the middleware only proved the token is valid, not the ability.
import (
"log"
"net/http"
"strconv"
"github.com/Molecule-AI/molecule-monorepo/platform/internal/db"
"github.com/Molecule-AI/molecule-monorepo/platform/internal/events"
"github.com/gin-gonic/gin"
)
// BroadcastHandler is constructed once and shared across requests.
type BroadcastHandler struct {
broadcaster *events.Broadcaster
}
// NewBroadcastHandler creates a BroadcastHandler.
func NewBroadcastHandler(b *events.Broadcaster) *BroadcastHandler {
return &BroadcastHandler{broadcaster: b}
}
// Broadcast handles POST /workspaces/:id/broadcast.
func (h *BroadcastHandler) Broadcast(c *gin.Context) {
senderID := c.Param("id")
if err := validateWorkspaceID(senderID); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "invalid workspace ID"})
return
}
var body struct {
Message string `json:"message" binding:"required"`
}
if err := c.ShouldBindJSON(&body); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "message is required"})
return
}
ctx := c.Request.Context()
// Verify sender exists and has broadcast_enabled=true.
var senderName string
var broadcastEnabled bool
err := db.DB.QueryRowContext(ctx,
`SELECT name, broadcast_enabled FROM workspaces WHERE id = $1 AND status != 'removed'`,
senderID,
).Scan(&senderName, &broadcastEnabled)
if err != nil {
c.JSON(http.StatusNotFound, gin.H{"error": "workspace not found"})
return
}
if !broadcastEnabled {
c.JSON(http.StatusForbidden, gin.H{
"error": "broadcast_disabled",
"hint": "This workspace does not have the broadcast ability. Ask a user or admin to enable it via PATCH /workspaces/:id/abilities.",
})
return
}
// Collect all non-removed agent workspaces (excludes the sender itself).
rows, err := db.DB.QueryContext(ctx,
`SELECT id FROM workspaces WHERE status != 'removed' AND id != $1`,
senderID,
)
if err != nil {
log.Printf("Broadcast: recipient query failed for %s: %v", senderID, err)
c.JSON(http.StatusInternalServerError, gin.H{"error": "internal error"})
return
}
defer rows.Close()
var recipientIDs []string
for rows.Next() {
var rid string
if rows.Scan(&rid) == nil {
recipientIDs = append(recipientIDs, rid)
}
}
if err := rows.Err(); err != nil {
log.Printf("Broadcast: recipient rows error for %s: %v", senderID, err)
c.JSON(http.StatusInternalServerError, gin.H{"error": "internal error"})
return
}
broadcastPayload := map[string]interface{}{
"message": body.Message,
"sender_id": senderID,
"sender": senderName,
}
// Persist broadcast_receive in each recipient's activity log + emit WS event.
delivered := 0
for _, rid := range recipientIDs {
if _, err := db.DB.ExecContext(ctx, `
INSERT INTO activity_logs (workspace_id, activity_type, method, source_id, summary, status)
VALUES ($1, 'broadcast_receive', 'broadcast', $2, $3, 'ok')
`, rid, senderID, "Broadcast from "+senderName+": "+broadcastTruncate(body.Message, 120)); err != nil {
log.Printf("Broadcast: activity_logs insert for recipient %s: %v", rid, err)
continue
}
h.broadcaster.BroadcastOnly(rid, "BROADCAST_MESSAGE", broadcastPayload)
delivered++
}
// Record the send on the sender's own log.
if _, err := db.DB.ExecContext(ctx, `
INSERT INTO activity_logs (workspace_id, activity_type, method, summary, status)
VALUES ($1, 'broadcast_sent', 'broadcast', $2, 'ok')
`, senderID, "Broadcast sent to "+strconv.Itoa(delivered)+" workspace(s)"); err != nil {
log.Printf("Broadcast: sender activity_log for %s: %v", senderID, err)
}
c.JSON(http.StatusOK, gin.H{
"status": "sent",
"delivered": delivered,
})
}
func broadcastTruncate(s string, max int) string {
runes := []rune(s)
if len(runes) <= max {
return s
}
return string(runes[:max]) + "…"
}
@@ -33,7 +33,6 @@ var wsColumns = []string{
"parent_id", "active_tasks", "max_concurrent_tasks", "last_error_rate", "last_sample_error",
"uptime_seconds", "current_task", "runtime", "workspace_dir", "x", "y", "collapsed",
"budget_limit", "monthly_spend",
"broadcast_enabled", "talk_to_user_enabled",
}
// ==================== GET — financial fields stripped from open endpoint ====================
@@ -53,10 +52,8 @@ func TestWorkspaceBudget_Get_NilLimit(t *testing.T) {
[]byte(`{}`), "http://localhost:9001",
nil, 0, 1, 0.0, "", 0, "", "langgraph", "",
0.0, 0.0, false,
nil, // budget_limit NULL
0, // monthly_spend 0
false, // broadcast_enabled
true)) // talk_to_user_enabled
nil, // budget_limit NULL
0)) // monthly_spend 0
w := httptest.NewRecorder()
c, _ := gin.CreateTestContext(w)
@@ -99,8 +96,7 @@ func TestWorkspaceBudget_Get_WithLimit(t *testing.T) {
nil, 0, 1, 0.0, "", 0, "", "langgraph", "",
0.0, 0.0, false,
int64(500), // budget_limit = $5.00 in DB
int64(123), // monthly_spend = $1.23 in DB
false, true)) // broadcast_enabled, talk_to_user_enabled
int64(123))) // monthly_spend = $1.23 in DB
w := httptest.NewRecorder()
c, _ := gin.CreateTestContext(w)
@@ -29,7 +29,6 @@ func TestWorkspaceGet_Success(t *testing.T) {
"parent_id", "active_tasks", "max_concurrent_tasks", "last_error_rate", "last_sample_error",
"uptime_seconds", "current_task", "runtime", "workspace_dir", "x", "y", "collapsed",
"budget_limit", "monthly_spend",
"broadcast_enabled", "talk_to_user_enabled",
}
mock.ExpectQuery("SELECT w.id, w.name").
WithArgs("cccccccc-0001-0000-0000-000000000000").
@@ -37,7 +36,7 @@ func TestWorkspaceGet_Success(t *testing.T) {
AddRow("cccccccc-0001-0000-0000-000000000000", "My Agent", "worker", 1, "online", []byte(`{"name":"test"}`),
"http://localhost:8001", nil, 2, 1, 0.05, "", 3600, "working", "langgraph",
"", 10.0, 20.0, false,
nil, 0, false, true))
nil, 0))
w := httptest.NewRecorder()
c, _ := gin.CreateTestContext(w)
@@ -119,7 +118,6 @@ func TestWorkspaceGet_RemovedReturns410(t *testing.T) {
"parent_id", "active_tasks", "max_concurrent_tasks", "last_error_rate", "last_sample_error",
"uptime_seconds", "current_task", "runtime", "workspace_dir", "x", "y", "collapsed",
"budget_limit", "monthly_spend",
"broadcast_enabled", "talk_to_user_enabled",
}
mock.ExpectQuery("SELECT w.id, w.name").
WithArgs(id).
@@ -127,7 +125,7 @@ func TestWorkspaceGet_RemovedReturns410(t *testing.T) {
AddRow(id, "Old Agent", "worker", 1, string(models.StatusRemoved), []byte(`null`),
"", nil, 0, 1, 0.0, "", 0, "", "langgraph",
"", 0.0, 0.0, false,
nil, 0, false, true))
nil, 0))
mock.ExpectQuery(`SELECT updated_at FROM workspaces`).
WithArgs(id).
WillReturnRows(sqlmock.NewRows([]string{"updated_at"}).AddRow(removedAt))
@@ -183,7 +181,6 @@ func TestWorkspaceGet_RemovedReturns410WithNullRemovedAtOnTimestampFetchFailure(
"parent_id", "active_tasks", "max_concurrent_tasks", "last_error_rate", "last_sample_error",
"uptime_seconds", "current_task", "runtime", "workspace_dir", "x", "y", "collapsed",
"budget_limit", "monthly_spend",
"broadcast_enabled", "talk_to_user_enabled",
}
mock.ExpectQuery("SELECT w.id, w.name").
WithArgs(id).
@@ -191,7 +188,7 @@ func TestWorkspaceGet_RemovedReturns410WithNullRemovedAtOnTimestampFetchFailure(
AddRow(id, "Vanished", "worker", 1, string(models.StatusRemoved), []byte(`null`),
"", nil, 0, 1, 0.0, "", 0, "", "langgraph",
"", 0.0, 0.0, false,
nil, 0, false, true))
nil, 0))
// Simulate the row vanishing between the two queries.
mock.ExpectQuery(`SELECT updated_at FROM workspaces`).
WithArgs(id).
@@ -246,7 +243,6 @@ func TestWorkspaceGet_RemovedWithIncludeQueryReturns200(t *testing.T) {
"parent_id", "active_tasks", "max_concurrent_tasks", "last_error_rate", "last_sample_error",
"uptime_seconds", "current_task", "runtime", "workspace_dir", "x", "y", "collapsed",
"budget_limit", "monthly_spend",
"broadcast_enabled", "talk_to_user_enabled",
}
mock.ExpectQuery("SELECT w.id, w.name").
WithArgs(id).
@@ -254,7 +250,7 @@ func TestWorkspaceGet_RemovedWithIncludeQueryReturns200(t *testing.T) {
AddRow(id, "Audit Agent", "worker", 1, string(models.StatusRemoved), []byte(`null`),
"", nil, 0, 1, 0.0, "", 0, "", "langgraph",
"", 0.0, 0.0, false,
nil, 0, false, true))
nil, 0))
// last_outbound_at follow-up query (existing path)
mock.ExpectQuery(`SELECT last_outbound_at FROM workspaces`).
WithArgs(id).
@@ -680,7 +676,6 @@ func TestWorkspaceList_Empty(t *testing.T) {
"parent_id", "active_tasks", "last_error_rate", "last_sample_error",
"uptime_seconds", "current_task", "runtime", "workspace_dir", "x", "y", "collapsed",
"budget_limit", "monthly_spend",
"broadcast_enabled", "talk_to_user_enabled",
}))
w := httptest.NewRecorder()
@@ -1384,7 +1379,6 @@ func TestWorkspaceGet_FinancialFieldsStripped(t *testing.T) {
"parent_id", "active_tasks", "max_concurrent_tasks", "last_error_rate", "last_sample_error",
"uptime_seconds", "current_task", "runtime", "workspace_dir", "x", "y", "collapsed",
"budget_limit", "monthly_spend",
"broadcast_enabled", "talk_to_user_enabled",
}
// Populate with non-zero financial values to confirm they are stripped.
mock.ExpectQuery("SELECT w.id, w.name").
@@ -1393,7 +1387,7 @@ func TestWorkspaceGet_FinancialFieldsStripped(t *testing.T) {
AddRow("cccccccc-0010-0000-0000-000000000000", "Finance Test", "worker", 1, "online", []byte(`{}`),
"http://localhost:9001", nil, 0, 1, 0.0, "", 0, "", "langgraph",
"", 0.0, 0.0, false,
int64(50000), int64(12500), false, true)) // budget_limit=500 USD, spend=125 USD
int64(50000), int64(12500))) // budget_limit=500 USD, spend=125 USD
w := httptest.NewRecorder()
c, _ := gin.CreateTestContext(w)
@@ -1441,7 +1435,6 @@ func TestWorkspaceGet_SensitiveFieldsStripped(t *testing.T) {
"parent_id", "active_tasks", "max_concurrent_tasks", "last_error_rate", "last_sample_error",
"uptime_seconds", "current_task", "runtime", "workspace_dir", "x", "y", "collapsed",
"budget_limit", "monthly_spend",
"broadcast_enabled", "talk_to_user_enabled",
}
mock.ExpectQuery("SELECT w.id, w.name").
WithArgs("cccccccc-0955-0000-0000-000000000000").
@@ -1454,7 +1447,7 @@ func TestWorkspaceGet_SensitiveFieldsStripped(t *testing.T) {
"langgraph",
"/home/user/secret-projects/client-work",
0.0, 0.0, false,
nil, 0, false, true))
nil, 0))
w := httptest.NewRecorder()
c, _ := gin.CreateTestContext(w)
@@ -36,15 +36,6 @@ type Workspace struct {
// to activity_logs, agent reads via GET /activity?since_id=). See
// migration 045 + RFC #2339.
DeliveryMode string `json:"delivery_mode" db:"delivery_mode"`
// BroadcastEnabled: when true the workspace may call POST /broadcast to
// deliver a message to all non-removed agent workspaces in the org.
// Default false — only privileged orchestrators should hold this ability.
BroadcastEnabled bool `json:"broadcast_enabled" db:"broadcast_enabled"`
// TalkToUserEnabled: when false the workspace's send_message_to_user calls
// and POST /notify requests are rejected with HTTP 403 so the agent is
// forced to route updates through a parent workspace. Default true
// (preserves existing behaviour for all workspaces).
TalkToUserEnabled bool `json:"talk_to_user_enabled" db:"talk_to_user_enabled"`
// Canvas layout fields (from JOIN)
X float64 `json:"x"`
Y float64 `json:"y"`
@@ -146,9 +146,6 @@ func Setup(hub *ws.Hub, broadcaster *events.Broadcaster, prov *provisioner.Provi
wsAdmin.GET("/workspaces", wh.List)
wsAdmin.POST("/workspaces", wh.Create)
wsAdmin.DELETE("/workspaces/:id", wh.Delete)
// Ability toggles — admin-only so workspace agents cannot self-modify
// broadcast_enabled or talk_to_user_enabled.
wsAdmin.PATCH("/workspaces/:id/abilities", handlers.PatchAbilities)
// Out-of-band bootstrap signal: CP's watcher POSTs here when it
// detects "RUNTIME CRASHED" in a workspace EC2 console output,
// so the canvas flips to failed in seconds instead of waiting
@@ -204,12 +201,6 @@ func Setup(hub *ws.Hub, broadcaster *events.Broadcaster, prov *provisioner.Provi
// to 'hibernated'. The workspace auto-wakes on the next A2A message.
wsAuth.POST("/hibernate", wh.Hibernate)
// Broadcast — send a message to all non-removed workspaces in the org.
// Requires broadcast_enabled=true on the source workspace (checked
// inside the handler). WorkspaceAuth on wsAuth proves token ownership.
broadcastH := handlers.NewBroadcastHandler(broadcaster)
wsAuth.POST("/broadcast", broadcastH.Broadcast)
// External-workspace credential lifecycle (issue #319 follow-up to
// the Create flow). Both endpoints reject runtime ≠ external with
// 400 — see external_rotate.go for the rationale.
@@ -1,3 +0,0 @@
ALTER TABLE workspaces
DROP COLUMN IF EXISTS broadcast_enabled,
DROP COLUMN IF EXISTS talk_to_user_enabled;
@@ -1,16 +0,0 @@
-- Workspace abilities: opt-in flags that gate platform-level behaviours.
--
-- broadcast_enabled (default FALSE): when TRUE the workspace may call
-- POST /workspaces/:id/broadcast to send a message to every non-removed
-- agent workspace in the org. Off by default — only privileged
-- orchestrator workspaces should hold this ability.
--
-- talk_to_user_enabled (default TRUE): when FALSE the workspace is not
-- allowed to deliver messages to the canvas user via send_message_to_user /
-- POST /notify. The platform returns HTTP 403 so the agent can forward its
-- update to a parent workspace instead. Default TRUE preserves existing
-- behaviour for all current workspaces.
ALTER TABLE workspaces
ADD COLUMN IF NOT EXISTS broadcast_enabled BOOLEAN NOT NULL DEFAULT FALSE,
ADD COLUMN IF NOT EXISTS talk_to_user_enabled BOOLEAN NOT NULL DEFAULT TRUE;
-6
View File
@@ -29,7 +29,6 @@ from typing import Callable
import inbox
from a2a_tools import (
tool_broadcast_message,
tool_chat_history,
tool_check_task_status,
tool_commit_memory,
@@ -161,11 +160,6 @@ async def handle_tool_call(name: str, arguments: dict) -> str:
arguments.get("before_ts", ""),
source_workspace_id=arguments.get("source_workspace_id") or None,
)
elif name == "broadcast_message":
return await tool_broadcast_message(
arguments.get("message", ""),
workspace_id=arguments.get("workspace_id") or None,
)
return f"Unknown tool: {name}"
-1
View File
@@ -137,7 +137,6 @@ from a2a_tools_delegation import ( # noqa: E402 (import after the from-a2a_cli
# identically.
from a2a_tools_messaging import ( # noqa: E402 (import after the top-of-module imports)
_upload_chat_files,
tool_broadcast_message,
tool_chat_history,
tool_get_workspace_info,
tool_list_peers,
-58
View File
@@ -101,50 +101,6 @@ async def _upload_chat_files(
return uploaded, None
async def tool_broadcast_message(
message: str,
workspace_id: str | None = None,
) -> str:
"""Send a broadcast message to ALL agent workspaces in the org.
Requires the workspace to have broadcast_enabled=true (set by a user or
admin via PATCH /workspaces/:id/abilities). Use for urgent org-wide
signals status changes, critical alerts, coordination instructions.
Every non-removed workspace receives the message in its activity log so
poll-mode agents pick it up, and push-mode canvases get a real-time
BROADCAST_MESSAGE WebSocket event.
Args:
message: The broadcast text. Keep it concise all agents receive
this, so avoid lengthy prose that floods every context.
workspace_id: Optional. Which registered workspace to send the
broadcast from. Single-workspace agents omit this.
"""
if not message:
return "Error: message is required"
target_workspace_id = (workspace_id or "").strip() or WORKSPACE_ID
try:
async with httpx.AsyncClient(timeout=30.0) as client:
resp = await client.post(
f"{PLATFORM_URL}/workspaces/{target_workspace_id}/broadcast",
json={"message": message},
headers=_auth_headers_for_heartbeat(target_workspace_id),
)
if resp.status_code == 200:
data = resp.json()
delivered = data.get("delivered", "?")
return f"Broadcast sent to {delivered} workspace(s)"
if resp.status_code == 403:
try:
hint = resp.json().get("hint", "")
except Exception:
hint = ""
return f"Error: broadcast ability not enabled.{(' ' + hint) if hint else ''}"
return f"Error: platform returned {resp.status_code}"
except Exception as e:
return f"Error sending broadcast: {e}"
async def tool_send_message_to_user(
message: str,
attachments: list[str] | None = None,
@@ -195,20 +151,6 @@ async def tool_send_message_to_user(
if uploaded:
return f"Message sent to user with {len(uploaded)} attachment(s)"
return "Message sent to user"
if resp.status_code == 403:
try:
body = resp.json()
if body.get("error") == "talk_to_user_disabled":
hint = body.get("hint", "")
return (
"Error: this workspace is not allowed to send messages "
"directly to the user (talk_to_user is disabled). "
+ (hint + " " if hint else "")
+ "Use delegate_task to forward your update to a parent "
"or supervisor workspace that can reach the user."
)
except Exception:
pass
return f"Error: platform returned {resp.status_code}"
except Exception as e:
return f"Error sending message: {e}"
-48
View File
@@ -3,57 +3,9 @@
import logging
import os
from abc import ABC, abstractmethod
from collections.abc import Mapping
from dataclasses import dataclass, field
from typing import Any
# ---------------------------------------------------------------------------
# Provider routing — type alias + resolver used by individual adapters.
# Each adapter defines its own ProviderRegistry with the providers it accepts.
# ---------------------------------------------------------------------------
# Maps prefix → (ordered_auth_env_vars, default_base_url).
ProviderRegistry = dict[str, tuple[tuple[str, ...], str]]
def resolve_provider_routing(
model_str: str,
env: Mapping[str, str],
*,
registry: ProviderRegistry,
runtime_config: dict[str, Any] | None = None,
) -> tuple[str, str, str]:
"""Resolve a ``provider:model`` string to ``(api_key, base_url, bare_model_id)``.
URL precedence (highest to lowest):
1. ``<PREFIX>_BASE_URL`` env var
2. ``runtime_config["provider_url"]``
3. registry default for the prefix
Unknown prefixes fall back to OPENAI_API_KEY + api.openai.com.
Raises RuntimeError when no API key env var is set for the prefix.
"""
if ":" in model_str:
prefix, model_id = model_str.split(":", 1)
else:
prefix, model_id = "openai", model_str
env_vars, default_url = registry.get(
prefix, (("OPENAI_API_KEY",), "https://api.openai.com/v1")
)
api_key = next((env[v] for v in env_vars if env.get(v)), "")
if not api_key:
raise RuntimeError(
f"No API key found for provider {prefix!r} "
f"(checked: {', '.join(env_vars)}). Set one in workspace secrets."
)
env_url = env.get(f"{prefix.upper()}_BASE_URL", "")
config_url = (runtime_config or {}).get("provider_url", "")
base_url = env_url or config_url or default_url
return api_key, base_url, model_id
from a2a.server.agent_execution import AgentExecutor
from event_log import DisabledEventLog, EventLogBackend
-4
View File
@@ -340,10 +340,6 @@ _CLI_A2A_COMMAND_KEYWORDS: dict[str, str | None] = {
"delegate_task_async": "delegate --async",
"check_task_status": "status",
"get_workspace_info": "info",
# `broadcast_message` is not exposed via the CLI subprocess interface
# today — it's an MCP-first capability. If a2a_cli grows a `broadcast`
# subcommand, map it here and the alignment test will gate the change.
"broadcast_message": None,
# `send_message_to_user` is not exposed via the CLI subprocess
# interface today — it requires a structured `attachments` field
# that wouldn't survive a positional-arg shell invocation cleanly.
-40
View File
@@ -51,7 +51,6 @@ from dataclasses import dataclass
from typing import Any, Literal
from a2a_tools import (
tool_broadcast_message,
tool_chat_history,
tool_check_task_status,
tool_commit_memory,
@@ -289,44 +288,6 @@ _GET_WORKSPACE_INFO = ToolSpec(
section=A2A_SECTION,
)
_BROADCAST_MESSAGE = ToolSpec(
name="broadcast_message",
short=(
"Send a message to ALL agent workspaces in the org simultaneously. "
"Requires broadcast_enabled=true on this workspace (set by user/admin)."
),
when_to_use=(
"Use for urgent, org-wide signals: critical status changes, emergency "
"stop instructions, coordinated task announcements. Every non-removed "
"workspace receives the message in its activity log (poll-mode agents "
"see it on their next poll; push-mode canvases get a real-time banner). "
"This tool returns an error if broadcast_enabled is false — a user or "
"admin must enable it via the workspace abilities settings first."
),
input_schema={
"type": "object",
"properties": {
"message": {
"type": "string",
"description": (
"The broadcast text. Keep it concise — every agent in the "
"org receives this in their activity feed."
),
},
"workspace_id": {
"type": "string",
"description": (
"Optional. Multi-workspace mode: the registered workspace "
"to broadcast from. Single-workspace agents omit this."
),
},
},
"required": ["message"],
},
impl=tool_broadcast_message,
section=A2A_SECTION,
)
_SEND_MESSAGE_TO_USER = ToolSpec(
name="send_message_to_user",
short=(
@@ -642,7 +603,6 @@ TOOLS: list[ToolSpec] = [
_CHECK_TASK_STATUS,
_LIST_PEERS,
_GET_WORKSPACE_INFO,
_BROADCAST_MESSAGE,
_SEND_MESSAGE_TO_USER,
# Inbox (standalone-only; in-container returns informational error)
_WAIT_FOR_MESSAGE,
@@ -5,7 +5,6 @@
- **check_task_status**: Poll the status of a task started with delegate_task_async; returns result when done.
- **list_peers**: List the workspaces this agent can communicate with — name, ID, status, role for each.
- **get_workspace_info**: Get this workspace's own info — ID, name, role, tier, parent, status.
- **broadcast_message**: Send a message to ALL agent workspaces in the org simultaneously. Requires broadcast_enabled=true on this workspace (set by user/admin).
- **send_message_to_user**: Send a message directly to the user's canvas chat — pushed instantly via WebSocket. Use this to: (1) acknowledge a task immediately ('Got it, I'll start working on this'), (2) send interim progress updates while doing long work, (3) deliver follow-up results after delegation completes, (4) attach files (zip, pdf, csv, image) for the user to download via the `attachments` field (NEVER paste file URLs in `message`). The message appears in the user's chat as if you're proactively reaching out.
- **wait_for_message**: Block until the next inbound message (canvas user OR peer agent) arrives, or until ``timeout_secs`` elapses.
- **inbox_peek**: List pending inbound messages without removing them.
@@ -27,9 +26,6 @@ Call this first when you need to delegate but don't know the target's ID. Access
### get_workspace_info
Use to introspect your own identity (e.g. before reporting back to the user, or to determine whether you're a tier-0 root that can write GLOBAL memory).
### broadcast_message
Use for urgent, org-wide signals: critical status changes, emergency stop instructions, coordinated task announcements. Every non-removed workspace receives the message in its activity log (poll-mode agents see it on their next poll; push-mode canvases get a real-time banner). This tool returns an error if broadcast_enabled is false — a user or admin must enable it via the workspace abilities settings first.
### send_message_to_user
Use proactively across the lifecycle of a task — early to acknowledge, mid-flight to update, late to deliver. Never paste file URLs in the message body — always pass absolute paths in `attachments` so the platform serves them as download chips (works on SaaS where external file hosts are unreachable).
+131 -119
View File
@@ -1,141 +1,153 @@
"""Unit tests for resolve_provider_routing in adapter_base.
"""Unit tests for OpenClaw adapter env-var key selection and provider URL routing.
Covers provider routing, URL-override precedence, and the missing-key error path.
Each adapter defines its own registry; this test file defines one inline that
mirrors what the openclaw adapter uses.
The key-selection and URL-routing logic lives inline in OpenClawAdapter.setup()
(adapter.py lines 84-92). Since setup() carries heavy subprocess dependencies,
these tests isolate the selection logic by reproducing the exact Python expressions
from the adapter source if the adapter's logic changes, these tests must be kept
in sync.
Organisation:
TestEnvKeyChain priority order of the 3 currently supported keys
TestProviderUrlMapping model-prefix provider URL dict correctness
TestNegativeAndFallback no keys set / unsupported keys
xfail stubs AISTUDIO + QIANFAN documented as not-yet-implemented
"""
from __future__ import annotations
import os
from unittest.mock import patch
import pytest
from adapter_base import ProviderRegistry, resolve_provider_routing
# Mirror of the registry in openclaw's adapter.py — kept in sync manually.
PROVIDER_REGISTRY: ProviderRegistry = {
"openai": (("OPENAI_API_KEY",), "https://api.openai.com/v1"),
"groq": (("GROQ_API_KEY",), "https://api.groq.com/openai/v1"),
"openrouter": (("OPENROUTER_API_KEY",), "https://openrouter.ai/api/v1"),
"qianfan": (("QIANFAN_API_KEY", "AISTUDIO_API_KEY"), "https://qianfan.baidubce.com/v2"),
"minimax": (("MINIMAX_API_KEY",), "https://api.minimaxi.com/v1"),
"moonshot": (("KIMI_API_KEY",), "https://api.moonshot.ai/v1"),
# ---------------------------------------------------------------------------
# Helpers — mirror the exact expressions from adapter.py lines 84-92.
# Must be kept in sync with the adapter source.
# ---------------------------------------------------------------------------
def _select_key(env: dict) -> str:
"""Mirror line 84: nested os.environ.get priority chain."""
return env.get("OPENAI_API_KEY",
env.get("GROQ_API_KEY",
env.get("OPENROUTER_API_KEY", "")))
_PROVIDER_URLS: dict[str, str] = {
"openai": "https://api.openai.com/v1",
"groq": "https://api.groq.com/openai/v1",
"openrouter": "https://openrouter.ai/api/v1",
}
class TestProviderRouting:
def test_openai_key_and_url(self):
api_key, base_url, model_id = resolve_provider_routing(
"openai:gpt-4o", {"OPENAI_API_KEY": "sk-openai"}, registry=PROVIDER_REGISTRY
)
assert api_key == "sk-openai"
assert base_url == "https://api.openai.com/v1"
assert model_id == "gpt-4o"
def test_groq_key_and_url(self):
api_key, base_url, model_id = resolve_provider_routing(
"groq:llama-3.3-70b", {"GROQ_API_KEY": "sk-groq"}, registry=PROVIDER_REGISTRY
)
assert api_key == "sk-groq"
assert base_url == "https://api.groq.com/openai/v1"
assert model_id == "llama-3.3-70b"
def test_openrouter_key_and_url(self):
api_key, base_url, model_id = resolve_provider_routing(
"openrouter:anthropic/claude-sonnet-4-5", {"OPENROUTER_API_KEY": "sk-or"}, registry=PROVIDER_REGISTRY
)
assert api_key == "sk-or"
assert base_url == "https://openrouter.ai/api/v1"
assert model_id == "anthropic/claude-sonnet-4-5"
def test_qianfan_primary_key(self):
api_key, _, _ = resolve_provider_routing(
"qianfan:ernie-4.5", {"QIANFAN_API_KEY": "sk-qf", "AISTUDIO_API_KEY": "sk-ai"}, registry=PROVIDER_REGISTRY
)
assert api_key == "sk-qf"
def test_qianfan_fallback_to_aistudio(self):
api_key, base_url, _ = resolve_provider_routing(
"qianfan:ernie-4.5", {"AISTUDIO_API_KEY": "sk-ai"}, registry=PROVIDER_REGISTRY
)
assert api_key == "sk-ai"
assert base_url == "https://qianfan.baidubce.com/v2"
def test_minimax_key_and_url(self):
api_key, base_url, model_id = resolve_provider_routing(
"minimax:MiniMax-M2.7", {"MINIMAX_API_KEY": "sk-mm"}, registry=PROVIDER_REGISTRY
)
assert api_key == "sk-mm"
assert base_url == "https://api.minimaxi.com/v1"
assert model_id == "MiniMax-M2.7"
def test_moonshot_key_and_url(self):
api_key, base_url, model_id = resolve_provider_routing(
"moonshot:kimi-k2.5", {"KIMI_API_KEY": "sk-kimi"}, registry=PROVIDER_REGISTRY
)
assert api_key == "sk-kimi"
assert base_url == "https://api.moonshot.ai/v1"
assert model_id == "kimi-k2.5"
def test_bare_model_id_defaults_to_openai(self):
api_key, base_url, model_id = resolve_provider_routing(
"gpt-4o", {"OPENAI_API_KEY": "sk-openai"}, registry=PROVIDER_REGISTRY
)
assert base_url == "https://api.openai.com/v1"
assert model_id == "gpt-4o"
def test_unknown_prefix_falls_back_to_openai_url(self):
api_key, base_url, model_id = resolve_provider_routing(
"custom-shim:my-model", {"OPENAI_API_KEY": "sk-openai"}, registry=PROVIDER_REGISTRY
)
assert base_url == "https://api.openai.com/v1"
assert model_id == "my-model"
def _select_url(model: str, runtime_config: dict | None = None) -> str:
"""Mirror lines 86-92: model-prefix → provider URL with optional override."""
prefix = model.split(":")[0] if ":" in model else "openai"
return (runtime_config or {}).get(
"provider_url",
_PROVIDER_URLS.get(prefix, "https://api.openai.com/v1"),
)
class TestUrlOverridePrecedence:
# ---------------------------------------------------------------------------
# 1. Env-var key priority chain (3 keys currently in adapter.py)
# ---------------------------------------------------------------------------
def test_env_base_url_beats_registry_default(self):
_, base_url, _ = resolve_provider_routing(
"minimax:MiniMax-M2.7",
{"MINIMAX_API_KEY": "sk-mm", "MINIMAX_BASE_URL": "https://api.minimax.chat/v1"},
registry=PROVIDER_REGISTRY,
)
assert base_url == "https://api.minimax.chat/v1"
class TestEnvKeyChain:
def test_runtime_config_provider_url_beats_registry_default(self):
_, base_url, _ = resolve_provider_routing(
"openai:gpt-4o",
{"OPENAI_API_KEY": "sk-openai"},
registry=PROVIDER_REGISTRY,
runtime_config={"provider_url": "https://proxy.example.com/v1"},
)
assert base_url == "https://proxy.example.com/v1"
def test_openai_key_selected(self):
with patch.dict(os.environ, {"OPENAI_API_KEY": "sk-openai-test"}, clear=True):
assert _select_key(os.environ) == "sk-openai-test"
def test_env_base_url_beats_runtime_config(self):
_, base_url, _ = resolve_provider_routing(
"openai:gpt-4o",
{"OPENAI_API_KEY": "sk-openai", "OPENAI_BASE_URL": "https://env-wins.com/v1"},
registry=PROVIDER_REGISTRY,
runtime_config={"provider_url": "https://config-loses.com/v1"},
)
assert base_url == "https://env-wins.com/v1"
def test_groq_key_selected_when_openai_absent(self):
with patch.dict(os.environ, {"GROQ_API_KEY": "sk-groq-test"}, clear=True):
assert _select_key(os.environ) == "sk-groq-test"
def test_openrouter_key_selected_when_openai_and_groq_absent(self):
with patch.dict(os.environ, {"OPENROUTER_API_KEY": "sk-or-test"}, clear=True):
assert _select_key(os.environ) == "sk-or-test"
def test_openai_beats_groq_when_both_set(self):
with patch.dict(os.environ, {"OPENAI_API_KEY": "openai", "GROQ_API_KEY": "groq"}, clear=True):
assert _select_key(os.environ) == "openai"
def test_groq_beats_openrouter_when_openai_absent(self):
with patch.dict(os.environ, {"GROQ_API_KEY": "groq", "OPENROUTER_API_KEY": "or"}, clear=True):
assert _select_key(os.environ) == "groq"
class TestMissingKey:
# ---------------------------------------------------------------------------
# 2. Model-prefix → provider URL routing
# ---------------------------------------------------------------------------
def test_raises_when_no_key_set(self):
with pytest.raises(RuntimeError, match="No API key found for provider 'minimax'"):
resolve_provider_routing("minimax:MiniMax-M2.7", {}, registry=PROVIDER_REGISTRY)
class TestProviderUrlMapping:
def test_raises_lists_checked_vars_in_message(self):
with pytest.raises(RuntimeError, match="MINIMAX_API_KEY"):
resolve_provider_routing("minimax:MiniMax-M2.7", {}, registry=PROVIDER_REGISTRY)
def test_openai_prefix_routes_to_openai(self):
assert _select_url("openai:gpt-4o") == "https://api.openai.com/v1"
def test_groq_prefix_routes_to_groq(self):
assert _select_url("groq:llama3-70b") == "https://api.groq.com/openai/v1"
def test_openrouter_prefix_routes_to_openrouter(self):
assert _select_url("openrouter:meta-llama/llama-3.3-70b") == "https://openrouter.ai/api/v1"
def test_runtime_config_override_wins_over_prefix(self):
url = _select_url("openai:gpt-4o", {"provider_url": "https://custom.example.com/v1"})
assert url == "https://custom.example.com/v1"
def test_unknown_prefix_falls_back_to_openai(self):
assert _select_url("some-unknown-model") == "https://api.openai.com/v1"
class TestRegistryCompleteness:
"""Smoke-check that every provider in the registry has a non-empty entry."""
# ---------------------------------------------------------------------------
# 3. Negative / fallback cases
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("prefix", PROVIDER_REGISTRY)
def test_all_providers_have_key_vars_and_url(self, prefix):
env_vars, base_url = PROVIDER_REGISTRY[prefix]
assert env_vars, f"{prefix}: env_vars is empty"
assert base_url.startswith("https://"), f"{prefix}: base_url looks wrong: {base_url}"
class TestNegativeAndFallback:
def test_no_keys_returns_empty_string(self):
with patch.dict(os.environ, {}, clear=True):
assert _select_key(os.environ) == ""
def test_unsupported_aistudio_key_returns_empty(self):
"""Documents that AISTUDIO_API_KEY is NOT yet in the adapter's key chain."""
with patch.dict(os.environ, {"AISTUDIO_API_KEY": "sk-ai"}, clear=True):
assert _select_key(os.environ) == ""
def test_unsupported_qianfan_key_returns_empty(self):
"""Documents that QIANFAN_API_KEY is NOT yet in the adapter's key chain."""
with patch.dict(os.environ, {"QIANFAN_API_KEY": "sk-qf"}, clear=True):
assert _select_key(os.environ) == ""
# ---------------------------------------------------------------------------
# 4. AISTUDIO + QIANFAN — xfail stubs (not yet implemented in adapter.py)
# These fail now; they should be promoted to passing tests once the adapter
# adds AISTUDIO_API_KEY and QIANFAN_API_KEY to its key chain and provider_urls.
# ---------------------------------------------------------------------------
@pytest.mark.xfail(
strict=True,
reason=(
"AISTUDIO_API_KEY not yet in openclaw adapter env-var chain — "
"add to adapter.py line 84 and provider_urls dict with "
"URL https://generativelanguage.googleapis.com/v1beta/openai"
),
)
def test_aistudio_key_routes_to_aistudio_url():
with patch.dict(os.environ, {"AISTUDIO_API_KEY": "sk-ai-test"}, clear=True):
assert _select_key(os.environ) == "sk-ai-test"
assert _select_url("gemini-2.5-flash") == "https://generativelanguage.googleapis.com/v1beta/openai"
@pytest.mark.xfail(
strict=True,
reason=(
"QIANFAN_API_KEY not yet in openclaw adapter env-var chain — "
"add to adapter.py line 84 and provider_urls dict with "
"URL https://qianfan.baidubce.com/v2"
),
)
def test_qianfan_key_routes_to_qianfan_url():
with patch.dict(os.environ, {"QIANFAN_API_KEY": "sk-qf-test"}, clear=True):
assert _select_key(os.environ) == "sk-qf-test"
assert _select_url("ernie-4.5") == "https://qianfan.baidubce.com/v2"