diff --git a/pkg/ai/provider/codex_appserver.go b/pkg/ai/provider/codex_appserver.go index 3bfe125..1ec17d7 100644 --- a/pkg/ai/provider/codex_appserver.go +++ b/pkg/ai/provider/codex_appserver.go @@ -516,7 +516,7 @@ func (c *CodexAppServer) handleNotification(method string, params json.RawMessag if terminal { ts.flushToolResults() } - if ev, ok := mapAppServerNotification(method, params, ctx); ok { + for _, ev := range mapAppServerNotification(method, params, ctx) { if ev.Kind == ai.EventResult && len(ts.outputSchema) > 0 { ev.StructuredData = json.RawMessage(ts.lastAgentMessage) } @@ -524,7 +524,7 @@ func (c *CodexAppServer) handleNotification(method string, params json.RawMessag it := parseAppServerNotif(params).Item if it != nil && it.Type == "commandExecution" { ts.queueToolResult(ev) - return + continue } } ts.send(ev) diff --git a/pkg/ai/provider/codex_appserver_lifecycle_ginkgo_test.go b/pkg/ai/provider/codex_appserver_lifecycle_ginkgo_test.go index 1c028fa..92623ac 100644 --- a/pkg/ai/provider/codex_appserver_lifecycle_ginkgo_test.go +++ b/pkg/ai/provider/codex_appserver_lifecycle_ginkgo_test.go @@ -20,6 +20,54 @@ func TestCodexAppServerLifecycle(t *testing.T) { } var _ = Describe("Codex app-server tool lifecycle", func() { + It("emits every operation from a dynamic exec call before its correlated result", func() { + const ( + sessionID = "thread-dynamic" + callID = "call-dynamic" + workdir = "/repo" + filePath = "pkg/cli/gitagent_e2e_test.go" + ) + script := `await tools.exec_command({cmd: "go test ./...", workdir: "/repo"}); +await tools.apply_patch({input: "*** Begin Patch\n*** Update File: pkg/cli/gitagent_e2e_test.go\n@@\n-old\n+new\n*** End Patch"});` + item := map[string]any{ + "id": callID, "type": "dynamicToolCall", "tool": "exec", + "arguments": script, "cwd": workdir, "status": "inProgress", + } + started, err := json.Marshal(map[string]any{"threadId": sessionID, "item": item}) + Expect(err).NotTo(HaveOccurred()) + + client, turn := activeGinkgoTurn() + client.handleNotification("item/started", started) + item["status"] = "completed" + item["success"] = true + completed, err := json.Marshal(map[string]any{"threadId": sessionID, "item": item}) + Expect(err).NotTo(HaveOccurred()) + client.handleNotification("item/completed", completed) + + events := drainEvents(turn) + Expect(events).To(HaveLen(3)) + Expect(events[0]).To(MatchFields(IgnoreExtras, Fields{ + "Kind": Equal(ai.EventToolUse), + "Tool": Equal("Bash"), + "Input": Equal(map[string]any{"command": "go test ./...", "input": script}), + "ToolCallID": Equal(callID), + "SessionID": Equal(sessionID), + })) + Expect(events[1]).To(MatchFields(IgnoreExtras, Fields{ + "Kind": Equal(ai.EventToolUse), + "Tool": Equal("Edit"), + "Input": HaveKeyWithValue("file_path", filePath), + "ToolCallID": Equal(callID + "#1"), + "SessionID": Equal(sessionID), + })) + Expect(events[2]).To(MatchFields(IgnoreExtras, Fields{ + "Kind": Equal(ai.EventToolResult), + "ToolCallID": Equal(callID), + "SessionID": Equal(sessionID), + "Success": BeTrue(), + })) + }) + It("emits one command use and one complete correlated result", func() { client, turn := activeGinkgoTurn() client.handleNotification("item/started", json.RawMessage(`{ diff --git a/pkg/ai/provider/codex_appserver_protocol.go b/pkg/ai/provider/codex_appserver_protocol.go index 7f3a7e0..731b341 100644 --- a/pkg/ai/provider/codex_appserver_protocol.go +++ b/pkg/ai/provider/codex_appserver_protocol.go @@ -123,57 +123,57 @@ func (n appServerNotif) foldUsage(usage *ai.Usage) { // --- notification mapping -------------------------------------------------- -// mapAppServerNotification maps one notification into an ai.Event. Pure: it -// folds thread/tokenUsage/updated into usage (ok=false) and reads the folded -// usage back out on turn/completed. +// mapAppServerNotification maps one notification into its ai.Events. Pure: it +// folds thread/tokenUsage/updated into usage without emitting and reads the +// folded usage back out on turn/completed. type appServerEventContext struct { Model string Usage *ai.Usage ToolOutput string } -func mapAppServerNotification(method string, params json.RawMessage, ctx appServerEventContext) (ai.Event, bool) { +func mapAppServerNotification(method string, params json.RawMessage, ctx appServerEventContext) []ai.Event { n := parseAppServerNotif(params) switch method { case "thread/started": sid := n.threadID() out := ai.Event{Kind: ai.EventSystem, Tool: "SessionInit", SessionID: sid, Model: ctx.Model} out.Raw = codexSessionToolUse(sid, ctx.Model) - return out, true + return []ai.Event{out} case "item/agentMessage/delta": if n.Delta == "" { - return ai.Event{}, false + return nil } - return ai.Event{Kind: ai.EventText, Text: n.Delta, SessionID: n.threadID(), Model: ctx.Model}, true + return []ai.Event{{Kind: ai.EventText, Text: n.Delta, SessionID: n.threadID(), Model: ctx.Model}} case "item/reasoning/textDelta", "item/reasoning/summaryTextDelta": if n.Delta == "" { - return ai.Event{}, false + return nil } - return ai.Event{Kind: ai.EventThinking, Text: n.Delta, SessionID: n.threadID(), Model: ctx.Model}, true + return []ai.Event{{Kind: ai.EventThinking, Text: n.Delta, SessionID: n.threadID(), Model: ctx.Model}} case "item/commandExecution/outputDelta": - return ai.Event{}, false + return nil case "item/started", "item/completed": return mapAppServerItem(method, n.Item, n.threadID(), ctx) case "thread/tokenUsage/updated": n.foldUsage(ctx.Usage) - return ai.Event{}, false + return nil case "turn/completed": if n.Turn != nil { switch n.Turn.Status { case "interrupted": - return ai.Event{}, false + return nil case "failed": message := "codex turn failed" if n.Turn.Error != nil { message = firstNonEmpty(n.Turn.Error.Message, n.Turn.Error.AdditionalDetails, message) } - return ai.Event{Kind: ai.EventError, Error: extractCodexErrorText(message), SessionID: n.threadID(), Model: ctx.Model}, true + return []ai.Event{{Kind: ai.EventError, Error: extractCodexErrorText(message), SessionID: n.threadID(), Model: ctx.Model}} } } out := ai.Event{Kind: ai.EventResult, Tool: "Result", SessionID: n.threadID(), Model: ctx.Model, Success: true} @@ -182,59 +182,72 @@ func mapAppServerNotification(method string, params json.RawMessage, ctx appServ out.Usage = &u } out.Raw = codexResultToolUse(out, n.ThreadID) - return out, true + return []ai.Event{out} case "turn/failed", "error": - return ai.Event{Kind: ai.EventError, Error: extractCodexErrorText(n.errorText()), SessionID: n.threadID(), Model: ctx.Model}, true + return []ai.Event{{Kind: ai.EventError, Error: extractCodexErrorText(n.errorText()), SessionID: n.threadID(), Model: ctx.Model}} } - return ai.Event{}, false + return nil } // mapAppServerItem dispatches item/started and item/completed on the item type: // agent messages become text, command/tool/file items become correlated use and // result events, and reasoning/user/hook items are dropped. -func mapAppServerItem(method string, it *appServerItemBody, sessionID string, ctx appServerEventContext) (ai.Event, bool) { +func mapAppServerItem(method string, it *appServerItemBody, sessionID string, ctx appServerEventContext) []ai.Event { if it == nil { - return ai.Event{}, false + return nil } switch it.Type { case "agentMessage", "plan": if method != "item/completed" || it.Text == "" { - return ai.Event{}, false + return nil } - return ai.Event{Kind: ai.EventText, Text: it.Text, SessionID: sessionID, Model: ctx.Model}, true + return []ai.Event{{Kind: ai.EventText, Text: it.Text, SessionID: sessionID, Model: ctx.Model}} case "reasoning", "userMessage", "hookPrompt", "": - return ai.Event{}, false + return nil } - use := history.NormalizeCodexToolCall(appServerToolCall(it, sessionID, ctx.Model)) if method == "item/started" { - out := ai.Event{ - Kind: ai.EventToolUse, Tool: use.Tool, Input: use.Input, - ToolCallID: use.ToolUseID, SessionID: sessionID, Model: ctx.Model, + uses := history.NormalizeCodexToolCalls(appServerToolCall(it, sessionID, ctx.Model)) + events := make([]ai.Event, 0, len(uses)) + for _, use := range uses { + out := ai.Event{ + Kind: ai.EventToolUse, Tool: use.Tool, Input: use.Input, + ToolCallID: use.ToolUseID, SessionID: sessionID, Model: ctx.Model, + } + out.Raw = codexToolUse(use, ctx.Model) + events = append(events, out) } - out.Raw = codexToolUse(use, ctx.Model) - return out, true + return events } if method != "item/completed" { - return ai.Event{}, false + return nil } + use := history.NormalizeCodexToolCall(appServerToolCall(it, sessionID, ctx.Model)) text := appServerToolResultText(it, ctx.ToolOutput) success := appServerToolSucceeded(it) use.Response = text raw := codexToolUse(use, ctx.Model) raw.IsError = !success - return ai.Event{ + return []ai.Event{{ Kind: ai.EventToolResult, Text: text, ToolCallID: use.ToolUseID, Success: success, SessionID: sessionID, Model: ctx.Model, Raw: raw, - }, true + }} } func appServerToolCall(it *appServerItemBody, sessionID, model string) history.CodexToolCall { name := it.Tool input := map[string]any{} + arguments := it.Arguments switch it.Type { case "commandExecution": name = "" + case "dynamicToolCall": + name = firstNonEmpty(name, it.Type) + var freeform string + if json.Unmarshal(it.Arguments, &freeform) == nil { + input["input"] = freeform + arguments = nil + } case "fileChange": name = "CodexPatchApply" if len(it.Changes) > 0 { @@ -246,7 +259,7 @@ func appServerToolCall(it *appServerItemBody, sessionID, model string) history.C name = firstNonEmpty(name, it.Type) } return history.CodexToolCall{ - Name: name, Namespace: it.Server, Arguments: it.Arguments, + Name: name, Namespace: it.Server, Arguments: arguments, Command: it.Command, Input: input, CWD: it.CWD, SessionID: sessionID, ID: it.ID, Model: model, } diff --git a/pkg/ai/provider/codex_appserver_test.go b/pkg/ai/provider/codex_appserver_test.go index 5aa0a16..3c638ab 100644 --- a/pkg/ai/provider/codex_appserver_test.go +++ b/pkg/ai/provider/codex_appserver_test.go @@ -178,14 +178,15 @@ func TestMapAppServerNotification_Kinds(t *testing.T) { } for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { - ev, ok := mapAppServerNotification(tc.method, json.RawMessage(tc.params), appServerEventContext{ + events := mapAppServerNotification(tc.method, json.RawMessage(tc.params), appServerEventContext{ Model: "gpt-5", Usage: &ai.Usage{}, }) if tc.drop { - assert.False(t, ok, "expected notification to be dropped, got %+v", ev) + assert.Empty(t, events, "expected notification to be dropped, got %+v", events) return } - require.True(t, ok, "expected an event for %s", tc.method) + require.Len(t, events, 1, "expected one event for %s", tc.method) + ev := events[0] assert.Equal(t, tc.want, ev.Kind) assert.Equal(t, "gpt-5", ev.Model) if tc.check != nil { @@ -340,15 +341,17 @@ func TestMapAppServerNotification_ErrorUnwrapping(t *testing.T) { nested := `The 'gpt-5.5-codex' model is not supported when using Codex with a ChatGPT account.` params := `{"threadId":"t","turnId":"u","willRetry":false,"error":{"message":"{\"type\":\"error\",\"status\":400,\"error\":{\"type\":\"invalid_request_error\",\"message\":\"` + nested + `\"}}"}}` - ev, ok := mapAppServerNotification("error", json.RawMessage(params), appServerEventContext{Model: "gpt-5", Usage: &ai.Usage{}}) - require.True(t, ok) + events := mapAppServerNotification("error", json.RawMessage(params), appServerEventContext{Model: "gpt-5", Usage: &ai.Usage{}}) + require.Len(t, events, 1) + ev := events[0] assert.Equal(t, ai.EventError, ev.Kind) assert.Equal(t, nested, ev.Error, "stringified upstream error payload should be unwrapped one level") } func TestMapAppServerNotification_TurnFailed(t *testing.T) { - ev, ok := mapAppServerNotification("turn/failed", json.RawMessage(`{"error":{"message":"boom"}}`), appServerEventContext{Model: "m", Usage: &ai.Usage{}}) - require.True(t, ok) + events := mapAppServerNotification("turn/failed", json.RawMessage(`{"error":{"message":"boom"}}`), appServerEventContext{Model: "m", Usage: &ai.Usage{}}) + require.Len(t, events, 1) + ev := events[0] assert.Equal(t, ai.EventError, ev.Kind) assert.Equal(t, "boom", ev.Error) } @@ -362,18 +365,19 @@ func TestMapAppServerNotification_UsageFolding(t *testing.T) { usage := &ai.Usage{} ctx := appServerEventContext{Model: "m", Usage: usage} - _, ok := mapAppServerNotification( + events := mapAppServerNotification( "thread/tokenUsage/updated", json.RawMessage(`{"tokenUsage":{"total":{"inputTokens":120,"outputTokens":40,"cachedInputTokens":12,"reasoningOutputTokens":7}}}`), ctx) - assert.False(t, ok, "token usage update emits no event") + assert.Empty(t, events, "token usage update emits no event") assert.Equal(t, 108, usage.InputTokens, "input net of cache (120-12)") assert.Equal(t, 33, usage.OutputTokens, "output net of reasoning (40-7)") assert.Equal(t, 12, usage.CacheReadTokens) assert.Equal(t, 7, usage.ReasoningTokens) - ev, ok := mapAppServerNotification("turn/completed", json.RawMessage(`{"threadId":"t","turn":{"id":"u"}}`), ctx) - require.True(t, ok) + events = mapAppServerNotification("turn/completed", json.RawMessage(`{"threadId":"t","turn":{"id":"u"}}`), ctx) + require.Len(t, events, 1) + ev := events[0] require.NotNil(t, ev.Usage, "turn/completed should carry the folded usage") assert.Equal(t, 108, ev.Usage.InputTokens) assert.Equal(t, 33, ev.Usage.OutputTokens)