diff --git a/internal/acp/execution.go b/internal/acp/execution.go index b15c335..6750b8b 100644 --- a/internal/acp/execution.go +++ b/internal/acp/execution.go @@ -153,14 +153,29 @@ func (s *Server) runSingleAgentViaExternalProvider( return nil, fmt.Errorf("external provider endpoint is missing") } forwardParams := sanitizeExternalACPParams(method, params) - return requestExternalACP( + collector := &externalACPNotificationCollector{} + combinedNotify := func(message map[string]any) { + collector.observe(message) + if notify != nil { + notify(message) + } + } + response, err := requestExternalACP( ctx, endpoint, provider.AuthorizationHeader, method, forwardParams, - notify, + combinedNotify, ) + if err != nil { + return nil, err + } + result := asMap(response["result"]) + if len(result) == 0 { + result = response + } + return collector.apply(result), nil } func resolveSingleAgentForwardEndpoint(provider syncedProvider) string { @@ -306,6 +321,80 @@ func normalizeAuthorizationHeader(raw string) string { return "Bearer " + normalized } +type externalACPNotificationCollector struct { + deltas strings.Builder + lastMessage string + turnID string + workingDirectory string +} + +func (c *externalACPNotificationCollector) observe(notification map[string]any) { + method := strings.TrimSpace(shared.StringArg(notification, "method", "")) + if method != "session.update" && method != "acp.session.update" { + return + } + params := asMap(notification["params"]) + if len(params) == 0 { + return + } + if turnID := strings.TrimSpace(shared.StringArg(params, "turnId", "")); turnID != "" { + c.turnID = turnID + } + for _, key := range []string{"resolvedWorkingDirectory", "effectiveWorkingDirectory", "workingDirectory"} { + if workingDirectory := strings.TrimSpace(shared.StringArg(params, key, "")); workingDirectory != "" { + c.workingDirectory = workingDirectory + break + } + } + if delta := strings.TrimSpace(shared.StringArg(params, "delta", "")); delta != "" { + if c.deltas.Len() > 0 { + c.deltas.WriteString("\n") + } + c.deltas.WriteString(delta) + } + message := strings.TrimSpace(shared.StringArg(params, "message", "")) + if message == "" { + message = strings.TrimSpace(shared.StringArg(asMap(params["message"]), "content", "")) + } + if message != "" && message != "session started" && message != "single-agent completed" { + c.lastMessage = message + } +} + +func (c *externalACPNotificationCollector) apply(result map[string]any) map[string]any { + if result == nil { + result = map[string]any{} + } + text := strings.TrimSpace(shared.StringArg(result, "output", "")) + if text == "" { + text = strings.TrimSpace(shared.StringArg(result, "summary", "")) + } + if text == "" { + text = strings.TrimSpace(shared.StringArg(result, "message", "")) + } + if text == "" { + text = strings.TrimSpace(c.deltas.String()) + } + if text == "" { + text = strings.TrimSpace(c.lastMessage) + } + if text != "" { + if _, exists := result["output"]; !exists { + result["output"] = text + } + if _, exists := result["summary"]; !exists { + result["summary"] = text + } + } + if _, exists := result["turnId"]; !exists && strings.TrimSpace(c.turnID) != "" { + result["turnId"] = strings.TrimSpace(c.turnID) + } + if _, exists := result["resolvedWorkingDirectory"]; !exists && strings.TrimSpace(c.workingDirectory) != "" { + result["resolvedWorkingDirectory"] = strings.TrimSpace(c.workingDirectory) + } + return result +} + func requestExternalACPWebSocket( ctx context.Context, endpoint *urlSpec, diff --git a/internal/acp/providers_sync_test.go b/internal/acp/providers_sync_test.go index 4cd7be1..f0438dc 100644 --- a/internal/acp/providers_sync_test.go +++ b/internal/acp/providers_sync_test.go @@ -296,3 +296,38 @@ func TestHandleRPCForwardsInboundBearerToExternalProvider(t *testing.T) { t.Fatalf("expected forwarded provider response, got %q", recorder.Body.String()) } } + +func TestExternalACPNotificationCollectorSynthesizesOutputAndWorkspace(t *testing.T) { + collector := &externalACPNotificationCollector{} + collector.observe(map[string]any{ + "jsonrpc": "2.0", + "method": "session.update", + "params": map[string]any{ + "sessionId": "session-streamed", + "threadId": "thread-streamed", + "turnId": "turn-streamed", + "type": "delta", + "delta": "streamed external output", + "resolvedWorkingDirectory": "/tmp/thread-streamed", + "pending": false, + "error": false, + }, + }) + + result := collector.apply(map[string]any{ + "success": true, + }) + + if got := result["output"]; got != "streamed external output" { + t.Fatalf("expected synthesized output from notifications, got %#v", result) + } + if got := result["summary"]; got != "streamed external output" { + t.Fatalf("expected synthesized summary from notifications, got %#v", result) + } + if got := result["turnId"]; got != "turn-streamed" { + t.Fatalf("expected synthesized turnId, got %#v", result) + } + if got := result["resolvedWorkingDirectory"]; got != "/tmp/thread-streamed" { + t.Fatalf("expected synthesized working directory, got %#v", result) + } +}