diff --git a/workspace-server/internal/handlers/activity.go b/workspace-server/internal/handlers/activity.go index c8dba3eaf..57adf78e9 100644 --- a/workspace-server/internal/handlers/activity.go +++ b/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 " 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) } } diff --git a/workspace-server/internal/handlers/activity_test.go b/workspace-server/internal/handlers/activity_test.go index 6cc4038f0..ec53a3f23 100644 --- a/workspace-server/internal/handlers/activity_test.go +++ b/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 " 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 ". + 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) + } +} diff --git a/workspace-server/internal/provisioner/isrunning_test.go b/workspace-server/internal/provisioner/isrunning_test.go index 0f5c587c0..3d217301e 100644 --- a/workspace-server/internal/provisioner/isrunning_test.go +++ b/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) + } + }) + } +} diff --git a/workspace-server/internal/provisioner/provisioner.go b/workspace-server/internal/provisioner/provisioner.go index e9171d778..dd94b08d3 100644 --- a/workspace-server/internal/provisioner/provisioner.go +++ b/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