fix(workspace-server): cascade-delete race + ACTIVITY_LOGGED body fidelity

workspace-server/internal/handlers/activity.go

@@ -519,6 +519,25 @@ func LogActivity(ctx context.Context, broadcaster events.EventEmitter, params Ac
 		if len(params.ToolTrace) > 0 {
 			payload["tool_trace"] = json.RawMessage(params.ToolTrace)
 		}
+		// Include request/response bodies in the live broadcast so the
+		// canvas's Agent Comms panel can render the actual task text
+		// and reply text immediately, instead of falling back to the
+		// "Delegating to <peer>" boilerplate. Without this, the live
+		// bubble was useless until a refresh re-fetched the activity
+		// row from /workspaces/:id/activity (which DOES return these
+		// columns from the DB). The workspace's report_activity helper
+		// caps each side at sensible sizes (4096 chars for error_detail,
+		// 256 for summary; request/response are bounded by the
+		// runtime's own caps — typical delegate_task payload is a few
+		// hundred chars to a few KB). json.RawMessage avoids a
+		// re-marshal round-trip; reqJSON/respJSON were already encoded
+		// for the DB insert above.
+		if reqStr != nil {
+			payload["request_body"] = json.RawMessage(reqJSON)
+		}
+		if respStr != nil {
+			payload["response_body"] = json.RawMessage(respJSON)
+		}
 		broadcaster.BroadcastOnly(params.WorkspaceID, "ACTIVITY_LOGGED", payload)
 	}
 }

workspace-server/internal/handlers/activity_test.go

@@ -2,6 +2,7 @@ package handlers
 
 import (
 	"bytes"
+	"context"
 	"database/sql/driver"
 	"encoding/json"
 	"fmt"
@@ -602,3 +603,126 @@ func TestScanSessionSearchRows_RowsErrPropagates(t *testing.T) {
 		t.Fatal("expected error to propagate")
 	}
 }
+
+// recordingBroadcaster records every BroadcastOnly invocation so a test
+// can assert what made it onto the wire. Implements events.EventEmitter.
+type recordingBroadcaster struct {
+	calls []recordedBroadcast
+}
+
+type recordedBroadcast struct {
+	workspaceID string
+	eventType   string
+	payload     map[string]interface{}
+}
+
+func (c *recordingBroadcaster) RecordAndBroadcast(_ context.Context, _ string, _ string, _ interface{}) error {
+	return nil
+}
+
+func (c *recordingBroadcaster) BroadcastOnly(workspaceID string, eventType string, payload interface{}) {
+	// Re-marshal/unmarshal so tests assert the actual wire shape (matches
+	// what hub.Broadcast does before sending). json.RawMessage values in
+	// the source payload survive the round-trip as their underlying JSON.
+	raw, err := json.Marshal(payload)
+	if err != nil {
+		c.calls = append(c.calls, recordedBroadcast{workspaceID, eventType, nil})
+		return
+	}
+	var out map[string]interface{}
+	if err := json.Unmarshal(raw, &out); err != nil {
+		c.calls = append(c.calls, recordedBroadcast{workspaceID, eventType, nil})
+		return
+	}
+	c.calls = append(c.calls, recordedBroadcast{workspaceID, eventType, out})
+}
+
+// TestLogActivity_Broadcast_IncludesRequestAndResponseBodies pins the
+// fix for the canvas Agent Comms "Delegating to <peer>" boilerplate
+// regression: without request_body/response_body in the live broadcast,
+// the panel renders the fallback string and the actual task text only
+// appears after a refresh re-fetches the row from /activity.
+func TestLogActivity_Broadcast_IncludesRequestAndResponseBodies(t *testing.T) {
+	mock := setupTestDB(t)
+	defer mock.ExpectationsWereMet()
+
+	mock.ExpectExec("INSERT INTO activity_logs").
+		WillReturnResult(sqlmock.NewResult(1, 1))
+
+	cb := &recordingBroadcaster{}
+	srcID := "ws-source"
+	tgtID := "ws-target"
+	method := "message/send"
+	summary := "Delegating to ws-target"
+	status := "ok"
+
+	LogActivity(context.Background(), cb, ActivityParams{
+		WorkspaceID:  "ws-source",
+		ActivityType: "a2a_send",
+		SourceID:     &srcID,
+		TargetID:     &tgtID,
+		Method:       &method,
+		Summary:      &summary,
+		RequestBody:  map[string]interface{}{"task": "audit moleculesai.app for performance"},
+		ResponseBody: nil,
+		Status:       status,
+	})
+
+	if len(cb.calls) != 1 {
+		t.Fatalf("expected 1 broadcast, got %d", len(cb.calls))
+	}
+	payload := cb.calls[0].payload
+	if payload["activity_type"] != "a2a_send" {
+		t.Errorf("activity_type missing/wrong: %v", payload["activity_type"])
+	}
+	// Critical: request_body must be present and carry the task text so
+	// the canvas's live-update path can render the actual delegation
+	// content instead of "Delegating to <peer>".
+	rb, ok := payload["request_body"].(map[string]interface{})
+	if !ok {
+		t.Fatalf("request_body missing from broadcast payload: got %#v", payload["request_body"])
+	}
+	if got := rb["task"]; got != "audit moleculesai.app for performance" {
+		t.Errorf("request_body.task = %v, want the actual task text", got)
+	}
+	// response_body was nil — must NOT be present (otherwise the canvas
+	// renders an empty agent reply bubble).
+	if _, present := payload["response_body"]; present {
+		t.Errorf("response_body should be omitted when nil, got %v", payload["response_body"])
+	}
+}
+
+func TestLogActivity_Broadcast_IncludesResponseBody(t *testing.T) {
+	mock := setupTestDB(t)
+	defer mock.ExpectationsWereMet()
+
+	mock.ExpectExec("INSERT INTO activity_logs").
+		WillReturnResult(sqlmock.NewResult(1, 1))
+
+	cb := &recordingBroadcaster{}
+	srcID := "ws-source"
+	method := "message/send"
+	status := "ok"
+
+	LogActivity(context.Background(), cb, ActivityParams{
+		WorkspaceID:  "ws-source",
+		ActivityType: "a2a_receive",
+		SourceID:     &srcID,
+		Method:       &method,
+		RequestBody:  map[string]interface{}{"task": "audit"},
+		ResponseBody: map[string]interface{}{"result": "LCP 2.1s, INP 180ms, CLS 0.05"},
+		Status:       status,
+	})
+
+	if len(cb.calls) != 1 {
+		t.Fatalf("expected 1 broadcast, got %d", len(cb.calls))
+	}
+	payload := cb.calls[0].payload
+	rb, ok := payload["response_body"].(map[string]interface{})
+	if !ok {
+		t.Fatalf("response_body missing from broadcast: got %#v", payload["response_body"])
+	}
+	if got := rb["result"]; got != "LCP 2.1s, INP 180ms, CLS 0.05" {
+		t.Errorf("response_body.result = %v, want the actual reply text", got)
+	}
+}

workspace-server/internal/provisioner/isrunning_test.go

@@ -49,3 +49,48 @@ func TestIsContainerNotFound(t *testing.T) {
 		})
 	}
 }
+
+// isRemovalInProgress decides whether Stop() treats Docker's "already
+// being removed" race as success (the container WILL be gone) versus
+// surfacing a 500 to the caller. False negative on cascade-delete
+// breaks the UX ("workspace marked removed, but stop call(s) failed —
+// please retry" when the workspace is, in fact, removed). False
+// positive would silently swallow a different daemon error and skip
+// the volume cleanup. Both directions matter — pin the truth table.
+func TestIsRemovalInProgress(t *testing.T) {
+	cases := []struct {
+		name string
+		err  error
+		want bool
+	}{
+		{"nil", nil, false},
+		{"docker race message",
+			errors.New(`Error response from daemon: removal of container ws-eeb99b5d-607 is already in progress`),
+			true},
+		{"docker race without ws prefix",
+			errors.New(`removal of container abc123 is already in progress`),
+			true},
+		// "already in progress" alone is too generic — would false-
+		// positive on e.g. "image pull is already in progress". Both
+		// substrings must be present.
+		{"unrelated already in progress",
+			errors.New(`image pull is already in progress`),
+			false},
+		{"not-found is NOT removal-in-progress",
+			errors.New(`Error response from daemon: No such container: ws-abc`),
+			false},
+		{"context deadline",
+			errors.New("context deadline exceeded"),
+			false},
+		{"empty string",
+			errors.New(""),
+			false},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := isRemovalInProgress(tc.err); got != tc.want {
+				t.Errorf("isRemovalInProgress(%q) = %v, want %v", tc.err, got, tc.want)
+			}
+		})
+	}
+}

workspace-server/internal/provisioner/provisioner.go

@@ -983,6 +983,17 @@ func (p *Provisioner) Stop(ctx context.Context, workspaceID string) error {
 		log.Printf("Provisioner: container %s already gone (no-op)", name)
 		return nil
 	}
+	if isRemovalInProgress(err) {
+		// Another concurrent caller (orphan sweeper, sibling cascade
+		// delete, manual `docker rm -f`) is already removing this
+		// container. The post-condition is the same as success: the
+		// container WILL be gone shortly. Surfacing this as a 500 on
+		// cascade-delete causes UI confusion ("workspace marked
+		// removed, but stop call(s) failed — please retry") even
+		// though retrying would just race the same in-flight removal.
+		log.Printf("Provisioner: container %s removal already in progress (no-op)", name)
+		return nil
+	}
 	// Real failure: daemon timeout, socket EOF, ctx cancellation, etc.
 	// Caller (workspace_crud.stopAndRemove, orphan_sweeper.sweepOnce)
 	// must propagate this so they can skip the follow-up RemoveVolume.
@@ -1048,6 +1059,29 @@ func isContainerNotFound(err error) bool {
 		strings.Contains(s, "not found")
 }
 
+// isRemovalInProgress detects the race where Docker is already removing
+// the container in response to a concurrent call. Symptom observed
+// during cascade-delete of a 7-workspace org: two of the seven returned
+//
+//	Error response from daemon: removal of container ws-xxx is already in progress
+//
+// because the platform's deletion fanout fired Stop() on every workspace
+// in parallel and the orphan sweeper happened to also reap two of them
+// at the same instant. The post-condition is identical to a successful
+// removal — the container WILL be gone — so callers should treat this
+// as a no-op rather than a real failure.
+//
+// String-match for the same reason as isContainerNotFound: docker/docker
+// surfaces this as a plain error string, no typed predicate. The CLI
+// itself relies on the message text.
+func isRemovalInProgress(err error) bool {
+	if err == nil {
+		return false
+	}
+	return strings.Contains(err.Error(), "removal of container") &&
+		strings.Contains(err.Error(), "already in progress")
+}
+
 // DockerClient returns the underlying Docker client for sharing with other handlers.
 func (p *Provisioner) DockerClient() *client.Client {
 	return p.cli