From 92d99d96fe645d3434e486bc69a757481101d26a Mon Sep 17 00:00:00 2001 From: Hongming Wang Date: Mon, 27 Apr 2026 13:25:32 -0700 Subject: [PATCH 1/2] fix(provisioner): treat "removal already in progress" as no-op success MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cascade-deleting a 7-workspace org returned 500 with "workspace marked removed, but 2 stop call(s) failed — please retry: stop eeb99b5d-...: force-remove ws-eeb99b5d-607: Error response from daemon: removal of container ws-eeb99b5d-607 is already in progress" even though the DB-side post-condition succeeded (removed_count=7) and the containers WERE removed shortly after. The fanout fired Stop() on every workspace concurrently and the orphan sweeper happened to reap two of them at the same instant, so Docker rejected the second ContainerRemove with "removal already in progress" — a race-condition ack, not a real failure. Retrying just races the same in-flight removal. The post-condition we care about (the container WILL be gone) is identical to a successful removal, so Stop() should treat it the same way it already treats "No such container" — a no-op return nil that lets the caller proceed with volume cleanup. Real daemon failures (timeout, EOF, ctx cancel) still surface as errors. Two pieces: - New isRemovalInProgress() predicate using the same string-match approach as isContainerNotFound (docker/docker has no typed errdef for this; the CLI itself relies on the message). - Stop() now treats the predicate as success, with a log line distinct from the not-found path so debugging can tell which race fired. Both substrings ("removal of container" + "already in progress") must match — "already in progress" alone would false-positive on unrelated operations like image pulls. Truth table pinned in 7 new test cases. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../internal/provisioner/isrunning_test.go | 45 +++++++++++++++++++ .../internal/provisioner/provisioner.go | 34 ++++++++++++++ 2 files changed, 79 insertions(+) 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 From c91c09dc55db8c2c7f862703a198270258431e78 Mon Sep 17 00:00:00 2001 From: Hongming Wang Date: Mon, 27 Apr 2026 13:38:23 -0700 Subject: [PATCH 2/2] fix(activity): include request/response bodies in ACTIVITY_LOGGED broadcast MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Canvas Agent Comms bubbles for outbound delegation showed only "Delegating to " boilerplate during the live update window — the actual task text only surfaced after a refresh re-fetched the row from /workspaces/:id/activity. Symptom flagged today during a fresh delegation manual test where the bubble said "Delegating to Perf Auditor" instead of the user's "audit moleculesai.app for performance" prompt. Root cause: LogActivity's broadcast payload at activity.go:510-518 deliberately omitted request_body and response_body, so the canvas's live-update path (AgentCommsPanel.tsx:271-289) saw `p.request_body = undefined` and toCommMessage fell back to the `Delegating to ${peerName}` template string. The DB row stored the real task / reply, which is why GET-on-mount worked. Fix: include both bodies in the broadcast as json.RawMessage values (no re-marshal cost — they were already encoded for the DB insert above). Same pattern as tool_trace, which has been included since #1814. Each side is bounded by the workspace-side caller's own caps: the runtime's report_activity helper caps error_detail at 4096 chars and summary at 256; request/response are constrained by the runtime's own limits — typical delegate_task payload is hundreds of chars to a few KB. If a much-larger broadcast becomes a concern later, a soft cap can be added at this site without breaking the contract. Two regression tests pin the broadcast shape: - request_body present → canvas renders the actual task text - response_body present → canvas renders the actual reply text - response_body nil → omitted from payload (no empty-bubble flicker) Co-Authored-By: Claude Opus 4.7 (1M context) --- .../internal/handlers/activity.go | 19 +++ .../internal/handlers/activity_test.go | 124 ++++++++++++++++++ 2 files changed, 143 insertions(+) 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) + } +}