diff --git a/internal/acp/execution_test.go b/internal/acp/execution_test.go index 4034e35..a955580 100644 --- a/internal/acp/execution_test.go +++ b/internal/acp/execution_test.go @@ -5,7 +5,10 @@ import ( "encoding/json" "net/http" "net/http/httptest" + "strings" "testing" + + "github.com/gorilla/websocket" ) func TestResolveSingleAgentForwardEndpointFromExampleConfig(t *testing.T) { @@ -176,6 +179,156 @@ func TestCodexCompatTranslatesSessionLifecycleToThreadAndTurnRPC(t *testing.T) { } } +func TestCodexCompatConvertsEmptyTurnResultToDisplayableFailure(t *testing.T) { + t.Parallel() + + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + defer func() { + _ = r.Body.Close() + }() + var request map[string]any + if err := json.NewDecoder(r.Body).Decode(&request); err != nil { + t.Fatalf("decode request: %v", err) + } + result := map[string]any{} + if stringValue(request["method"]) == "thread/start" { + result["id"] = "codex-thread-1" + } + _ = json.NewEncoder(w).Encode(map[string]any{ + "jsonrpc": "2.0", + "id": request["id"], + "result": result, + }) + })) + defer upstream.Close() + + compat := newProviderCompat(syncedProvider{ + ProviderID: "codex", + Label: "Codex", + Endpoint: upstream.URL, + Enabled: true, + }) + result, err := compat.StartSession( + context.Background(), + "session-1", + "thread-1", + map[string]any{ + "taskPrompt": "Reply with exactly pong", + "workingDirectory": t.TempDir(), + }, + nil, + ) + if err != nil { + t.Fatalf("StartSession failed: %v", err) + } + if got := result["success"]; got != false { + t.Fatalf("expected failure success flag, got %#v", result) + } + if got := result["error"]; got != "codex returned no displayable output" { + t.Fatalf("expected displayable error, got %#v", result) + } +} + +func TestCodexCompatWaitsForTurnCompletedNotification(t *testing.T) { + t.Parallel() + + upgrader := websocket.Upgrader{} + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := upgrader.Upgrade(w, r, nil) + if err != nil { + t.Fatalf("upgrade websocket: %v", err) + } + defer func() { + _ = conn.Close() + }() + for { + var request map[string]any + if err := conn.ReadJSON(&request); err != nil { + return + } + method := stringValue(request["method"]) + switch method { + case "initialize": + if err := conn.WriteJSON(map[string]any{ + "jsonrpc": "2.0", + "id": request["id"], + "result": map[string]any{"protocolVersion": 1}, + }); err != nil { + t.Fatalf("write initialize response: %v", err) + } + case "thread/start": + if err := conn.WriteJSON(map[string]any{ + "jsonrpc": "2.0", + "id": request["id"], + "result": map[string]any{"id": "codex-thread-1"}, + }); err != nil { + t.Fatalf("write thread response: %v", err) + } + case "turn/start": + turn := map[string]any{ + "id": "turn-1", + "status": "inProgress", + "items": []any{}, + } + if err := conn.WriteJSON(map[string]any{ + "jsonrpc": "2.0", + "id": request["id"], + "result": map[string]any{"turn": turn}, + }); err != nil { + t.Fatalf("write turn response: %v", err) + } + if err := conn.WriteJSON(map[string]any{ + "method": "item/completed", + "params": map[string]any{ + "item": map[string]any{ + "type": "assistant_message", + "content": []any{map[string]any{"text": "pong"}}, + }, + }, + }); err != nil { + t.Fatalf("write item completed: %v", err) + } + turn["status"] = "completed" + if err := conn.WriteJSON(map[string]any{ + "method": "turn/completed", + "params": map[string]any{ + "threadId": "codex-thread-1", + "turn": turn, + }, + }); err != nil { + t.Fatalf("write turn completed: %v", err) + } + default: + t.Fatalf("unexpected method %q", method) + } + } + })) + defer upstream.Close() + + compat := newProviderCompat(syncedProvider{ + ProviderID: "codex", + Label: "Codex", + Endpoint: "ws" + strings.TrimPrefix(upstream.URL, "http"), + Enabled: true, + }) + result, err := compat.StartSession( + context.Background(), + "session-1", + "thread-1", + map[string]any{ + "taskPrompt": "Reply with exactly pong", + "workingDirectory": t.TempDir(), + }, + nil, + ) + if err != nil { + t.Fatalf("StartSession failed: %v", err) + } + if got := result["output"]; got != "pong" { + t.Fatalf("expected output pong after turn/completed, got %#v", result) + } +} + func TestExternalACPNotificationCollectorExtractsNestedSessionUpdateText(t *testing.T) { t.Parallel() @@ -202,6 +355,33 @@ func TestExternalACPNotificationCollectorExtractsNestedSessionUpdateText(t *test } } +func TestExternalACPNotificationCollectorConvertsToolErrorToFailure(t *testing.T) { + t.Parallel() + + collector := &externalACPNotificationCollector{} + collector.observe(map[string]any{ + "method": "session.update", + "params": map[string]any{ + "update": map[string]any{ + "sessionUpdate": "tool_error", + "error": true, + "message": "exec_command failed: Failed to create unified exec process", + }, + }, + }) + + result := collector.apply(map[string]any{}) + if got := result["success"]; got != false { + t.Fatalf("expected failure result, got %#v", result) + } + if got := result["error"]; got != "exec_command failed: Failed to create unified exec process" { + t.Fatalf("expected tool error text, got %#v", result) + } + if _, ok := result["output"]; ok { + t.Fatalf("did not expect tool error to become output, got %#v", result) + } +} + func TestExternalACPNotificationCollectorPrefersStreamTextOverAckResult(t *testing.T) { t.Parallel() diff --git a/internal/acp/helpers.go b/internal/acp/helpers.go index 99f80d3..df2555a 100644 --- a/internal/acp/helpers.go +++ b/internal/acp/helpers.go @@ -69,6 +69,7 @@ func normalizeAuthorizationHeader(raw string) string { type externalACPNotificationCollector struct { deltas strings.Builder lastMessage string + errors []string turnID string } @@ -84,10 +85,20 @@ func (c *externalACPNotificationCollector) observe(notification map[string]any) if turnID := strings.TrimSpace(stringValue(params["turnId"])); turnID != "" { c.turnID = turnID } + if errorText := extractExternalACPNotificationError(notification); errorText != "" { + c.errors = append(c.errors, errorText) + } + if strings.TrimSpace(stringValue(notification["method"])) == "turn/completed" { + return + } updateText := extractExternalACPNotificationText(notification) if updateText == "" { return } + if isExternalACPFailureText(updateText) { + c.errors = append(c.errors, updateText) + return + } if c.deltas.Len() > 0 { c.deltas.WriteString("\n") } @@ -122,7 +133,13 @@ func (c *externalACPNotificationCollector) apply(result map[string]any) map[stri break } } - if text != "" { + if errorText := c.errorText(); errorText != "" { + result["success"] = false + result["error"] = errorText + result["message"] = errorText + delete(result, "output") + delete(result, "summary") + } else if text != "" { result["output"] = text result["summary"] = text } @@ -132,6 +149,26 @@ func (c *externalACPNotificationCollector) apply(result map[string]any) map[stri return result } +func (c *externalACPNotificationCollector) errorText() string { + if c == nil || len(c.errors) == 0 { + return "" + } + seen := make(map[string]struct{}, len(c.errors)) + var parts []string + for _, item := range c.errors { + text := strings.TrimSpace(item) + if text == "" { + continue + } + if _, ok := seen[text]; ok { + continue + } + seen[text] = struct{}{} + parts = append(parts, text) + } + return strings.TrimSpace(strings.Join(parts, "\n")) +} + func isGenericHermesAckText(text string) bool { switch strings.ToLower(strings.TrimSpace(text)) { case "", "ok", "session started", "single-agent completed": @@ -145,6 +182,9 @@ func extractExternalACPNotificationText(notification map[string]any) string { if notification == nil { return "" } + if strings.TrimSpace(stringValue(notification["method"])) == "turn/completed" { + return "" + } payload := asMap(notification["params"]) if len(payload) == 0 { payload = notification @@ -163,8 +203,11 @@ func extractExternalACPNotificationText(notification map[string]any) string { if text := extractExternalACPTextValue(update); text != "" { return text } - if text := extractExternalACPTextValue(asMap(payload["item"])); text != "" { - return text + item := asMap(payload["item"]) + if strings.TrimSpace(stringValue(item["type"])) != "userMessage" { + if text := extractExternalACPTextValue(item); text != "" { + return text + } } if text := extractExternalACPTextValue(payload); text != "" { return text @@ -172,6 +215,60 @@ func extractExternalACPNotificationText(notification map[string]any) string { return "" } +func extractExternalACPNotificationError(notification map[string]any) string { + if notification == nil { + return "" + } + payload := asMap(notification["params"]) + if len(payload) == 0 { + payload = notification + } + update := asMap(payload["update"]) + if len(update) == 0 { + update = payload + } + if turnError := extractExternalACPTextValue(asMap(asMap(payload["turn"])["error"])); turnError != "" { + return turnError + } + for _, source := range []map[string]any{update, asMap(payload["item"]), payload} { + if len(source) == 0 { + continue + } + if !parseBool(source["error"]) && strings.TrimSpace(stringValue(source["level"])) != "error" { + if text := extractExternalACPTextValue(source); !isExternalACPFailureText(text) { + continue + } + } + for _, key := range []string{"error", "message", "text", "content", "delta", "value"} { + if text := extractExternalACPTextValue(source[key]); text != "" { + return text + } + } + if text := extractExternalACPTextValue(source); text != "" { + return text + } + } + return "" +} + +func isExternalACPFailureText(text string) bool { + normalized := strings.ToLower(strings.TrimSpace(text)) + if normalized == "" { + return false + } + for _, marker := range []string{ + "exec_command failed", + "failed to create unified exec process", + "execution_failed", + "tool execution failed", + } { + if strings.Contains(normalized, marker) { + return true + } + } + return false +} + func extractExternalACPTextValue(value any) string { switch v := value.(type) { case string: diff --git a/internal/acp/orchestrator.go b/internal/acp/orchestrator.go index b46f923..0452ce8 100644 --- a/internal/acp/orchestrator.go +++ b/internal/acp/orchestrator.go @@ -190,11 +190,14 @@ func (o *SessionOrchestrator) normalizeResult(sess *session, result map[string]a result = map[string]any{} } + successValue, hasSuccess := result["success"] + success := !hasSuccess || parseBool(successValue) + output := strings.TrimSpace(shared.StringArg(result, "output", "")) if output == "" { output = strings.TrimSpace(shared.StringArg(result, "summary", "")) } - if output == "" { + if output == "" && success { output = strings.TrimSpace(shared.StringArg(result, "message", "")) } @@ -206,7 +209,9 @@ func (o *SessionOrchestrator) normalizeResult(sess *session, result map[string]a result["turnId"] = turnID result["status"] = "completed" - result["success"] = true + if !hasSuccess { + result["success"] = true + } result["resolvedExecutionTarget"] = routing.TargetID result["resolvedProviderId"] = routing.ProviderID result["resolvedGatewayProviderId"] = routing.GatewayProviderID @@ -218,6 +223,14 @@ func (o *SessionOrchestrator) normalizeResult(sess *session, result map[string]a result["summary"] = output } } + if output == "" && routing.TargetID != "gateway" && !parseBool(result["success"]) { + result["status"] = "failed" + } else if output == "" && routing.TargetID != "gateway" { + result["success"] = false + result["status"] = "failed" + result["error"] = "provider returned no displayable output" + result["message"] = "provider returned no displayable output" + } workingDirectory := shared.StringArg(params, "workingDirectory", "") routingParams := shared.AsMap(params["routing"]) diff --git a/internal/acp/provider_compat.go b/internal/acp/provider_compat.go index 6441b9a..7d68adc 100644 --- a/internal/acp/provider_compat.go +++ b/internal/acp/provider_compat.go @@ -101,6 +101,9 @@ func (c *codexCompat) Probe(ctx context.Context) ProviderProbeResult { } func (c *codexCompat) StartSession(ctx context.Context, sessionID string, threadID string, params map[string]any, sink SessionNotificationSink) (map[string]any, error) { + if c.transport() == "ws" { + return c.startSessionWS(ctx, sessionID, threadID, params, sink) + } thread, err := c.codexCall(ctx, "thread/start", codexThreadStartParams(params), nil) if err != nil { return nil, err @@ -114,6 +117,9 @@ func (c *codexCompat) StartSession(ctx context.Context, sessionID string, thread } func (c *codexCompat) SendMessage(ctx context.Context, sessionID string, threadID string, params map[string]any, sink SessionNotificationSink) (map[string]any, error) { + if c.transport() == "ws" { + return c.sendMessageWS(ctx, sessionID, threadID, params, sink) + } codexThreadID := c.lookupThread(sessionID, threadID) if codexThreadID == "" { codexThreadID = strings.TrimSpace(threadID) @@ -134,6 +140,42 @@ func (c *codexCompat) SendMessage(ctx context.Context, sessionID string, threadI return c.startTurn(ctx, codexThreadID, params, sink) } +func (c *codexCompat) startSessionWS(ctx context.Context, sessionID string, threadID string, params map[string]any, sink SessionNotificationSink) (map[string]any, error) { + return c.withInitializedCodexWS(ctx, func(conn *websocket.Conn) (map[string]any, error) { + thread, err := c.writeAndReadWSRPC(ctx, conn, "thread/start", codexThreadStartParams(params), nil) + if err != nil { + return nil, err + } + codexThreadID := codexThreadIDFromResult(thread) + if codexThreadID == "" { + return nil, fmt.Errorf("codex thread/start response missing thread id") + } + c.rememberThread(sessionID, threadID, codexThreadID) + return c.startTurnOnWS(ctx, conn, codexThreadID, params, sink) + }) +} + +func (c *codexCompat) sendMessageWS(ctx context.Context, sessionID string, threadID string, params map[string]any, sink SessionNotificationSink) (map[string]any, error) { + codexThreadID := c.lookupThread(sessionID, threadID) + if codexThreadID == "" { + codexThreadID = strings.TrimSpace(threadID) + } + if codexThreadID == "" { + return c.startSessionWS(ctx, sessionID, threadID, params, sink) + } + return c.withInitializedCodexWS(ctx, func(conn *websocket.Conn) (map[string]any, error) { + thread, err := c.writeAndReadWSRPC(ctx, conn, "thread/resume", map[string]any{"threadId": codexThreadID}, nil) + if err != nil { + return nil, err + } + if resolved := codexThreadIDFromResult(thread); resolved != "" { + codexThreadID = resolved + c.rememberThread(sessionID, threadID, codexThreadID) + } + return c.startTurnOnWS(ctx, conn, codexThreadID, params, sink) + }) +} + func (c *codexCompat) CloseSession(ctx context.Context, sessionID string) error { c.mu.Lock() delete(c.threads, sessionID) @@ -163,13 +205,63 @@ func (c *codexCompat) startTurn(ctx context.Context, codexThreadID string, param if err != nil { return nil, err } + return c.finalizeCodexTurnResult(codexThreadID, result), nil +} + +func (c *codexCompat) startTurnOnWS(ctx context.Context, conn *websocket.Conn, codexThreadID string, params map[string]any, sink SessionNotificationSink) (map[string]any, error) { + result, err := c.writeAndReadWSRPC( + ctx, + conn, + "turn/start", + map[string]any{ + "threadId": codexThreadID, + "input": codexUserInput(params), + }, + sink, + ) + if err != nil { + return nil, err + } + return c.finalizeCodexTurnResult(codexThreadID, result), nil +} + +func (c *codexCompat) finalizeCodexTurnResult(codexThreadID string, result map[string]any) map[string]any { if _, ok := result["output"]; !ok { if summary := strings.TrimSpace(shared.StringArg(result, "summary", "")); summary != "" { result["output"] = summary } } result["providerThreadId"] = codexThreadID - return result, nil + if codexDisplayText(result) == "" && !isProviderFailureResult(result) { + result["success"] = false + result["error"] = "codex returned no displayable output" + result["message"] = "codex returned no displayable output" + } + return result +} + +func codexDisplayText(result map[string]any) string { + for _, key := range []string{"output", "summary", "message"} { + if text := strings.TrimSpace(shared.StringArg(result, key, "")); text != "" && !isGenericHermesAckText(text) { + return text + } + } + return "" +} + +func isProviderFailureResult(result map[string]any) bool { + if result == nil { + return false + } + if value, ok := result["success"]; ok && !parseBool(value) { + return true + } + for _, key := range []string{"error", "errorMessage", "unavailableMessage"} { + if strings.TrimSpace(shared.StringArg(result, key, "")) != "" { + return true + } + } + return false } func (c *codexCompat) codexCall(ctx context.Context, method string, params map[string]any, sink SessionNotificationSink) (map[string]any, error) { @@ -180,6 +272,12 @@ func (c *codexCompat) codexCall(ctx context.Context, method string, params map[s } func (c *codexCompat) callWSRPCWithInitialize(ctx context.Context, method string, params map[string]any, sink SessionNotificationSink) (map[string]any, error) { + return c.withInitializedCodexWS(ctx, func(conn *websocket.Conn) (map[string]any, error) { + return c.writeAndReadWSRPC(ctx, conn, method, params, sink) + }) +} + +func (c *codexCompat) withInitializedCodexWS(ctx context.Context, run func(*websocket.Conn) (map[string]any, error)) (map[string]any, error) { headers := http.Header{} if c.authHeader != "" { headers.Set("Authorization", c.authHeader) @@ -193,7 +291,7 @@ func (c *codexCompat) callWSRPCWithInitialize(ctx context.Context, method string if _, err := c.writeAndReadWSRPC(ctx, conn, "initialize", codexInitializeParams(), nil); err != nil { return nil, err } - return c.writeAndReadWSRPC(ctx, conn, method, params, sink) + return run(conn) } func (c *codexCompat) writeAndReadWSRPC(ctx context.Context, conn *websocket.Conn, method string, params map[string]any, sink SessionNotificationSink) (map[string]any, error) { @@ -209,12 +307,14 @@ func (c *codexCompat) writeAndReadWSRPC(ctx context.Context, conn *websocket.Con } collector := &externalACPNotificationCollector{} + var pendingTurn map[string]any for { select { case <-ctx.Done(): return nil, ctx.Err() default: } + _ = conn.SetReadDeadline(time.Now().Add(5 * time.Minute)) _, payload, err := conn.ReadMessage() if err != nil { return nil, err @@ -234,6 +334,9 @@ func (c *codexCompat) writeAndReadWSRPC(ctx context.Context, conn *websocket.Con sink(update) } } + if pendingTurn != nil && isCodexTurnCompletedNotification(decoded, pendingTurn) { + return collector.apply(pendingTurn), nil + } continue } @@ -245,10 +348,48 @@ func (c *codexCompat) writeAndReadWSRPC(ctx context.Context, conn *websocket.Con if err != nil { return nil, err } + if method == "turn/start" && isCodexTurnInProgress(result) { + pendingTurn = collector.apply(result) + continue + } return collector.apply(result), nil } } +func isCodexTurnInProgress(result map[string]any) bool { + if result == nil { + return false + } + turn := shared.AsMap(result["turn"]) + if len(turn) == 0 { + return false + } + status := strings.TrimSpace(shared.StringArg(turn, "status", "")) + return status == "" || strings.EqualFold(status, "inProgress") || strings.EqualFold(status, "running") +} + +func isCodexTurnCompletedNotification(notification map[string]any, pendingTurn map[string]any) bool { + if notification == nil || pendingTurn == nil { + return false + } + if strings.TrimSpace(shared.StringArg(notification, "method", "")) != "turn/completed" { + return false + } + params := shared.AsMap(notification["params"]) + turn := shared.AsMap(params["turn"]) + if len(turn) == 0 { + return true + } + pending := shared.AsMap(pendingTurn["turn"]) + pendingID := strings.TrimSpace(shared.StringArg(pending, "id", "")) + completedID := strings.TrimSpace(shared.StringArg(turn, "id", "")) + if pendingID == "" || completedID == "" || pendingID == completedID { + pendingTurn["turn"] = turn + return true + } + return false +} + func (c *codexCompat) rememberThread(sessionID string, threadID string, codexThreadID string) { c.mu.Lock() defer c.mu.Unlock()