package agentruntime import ( "log/slog" "mime" "path/filepath" "strings" activityshared "github.com/tutti-os/tutti/packages/agent/daemon/activity/events" ) func appServerCompactionNoticeEvent(session Session, turnID string, messageID string, status string) activityshared.Event { title := appServerCompactionInterruptedTitle switch status { case "running": title = appServerCompactingContextTitle case "completed": title = appServerContextCompactedTitle } return appServerSystemNoticeEvent(session, turnID, "system_notice", title, "", map[string]any{ "messageId": messageID, "noticeCommand": "compact", "noticeCommandStatus": status, }) } func (*CodexAppServerAdapter) appServerItemEvents( session Session, turnID string, item map[string]any, completed bool, normalizer *acpTurnNormalizer, ) []activityshared.Event { if len(item) == 0 || normalizer == nil { return nil } itemType := asString(item["type"]) // Review/compaction items stream both item/started and item/completed. // Gate each review banner to a single lifecycle event so the GUI shows it // once; compaction emits on both so the GUI can show live progress. if notice, ok := appServerNoticeItems[itemType]; ok { if notice.emitOnCompleted != completed { return nil } return []activityshared.Event{appServerSystemNoticeEvent(session, turnID, "system_notice", notice.message, "")} } if itemType == "contextCompaction" { // appServerSlashCompact emits the "Compacting context." banner eagerly // (before this notification can even arrive) and tracks its messageId // on the normalizer up front so the transcript row exists even if Codex // app-server never streams item/started at all. When that's already // the case, reuse the normalizer's stable messageId instead of deriving // a new one from the item id: otherwise item/started would append a // second, unrelated banner row rather than confirming the one already // shown. messageID := "compaction:" + firstNonEmpty(asString(item["id"]), turnID) shouldEmit := false if completed { messageID, shouldEmit = normalizer.CompleteCompactionNotice(messageID) } else { messageID, shouldEmit = normalizer.StartCompactionNotice(messageID) } if !shouldEmit { return nil } status := "running" if completed { status = "completed" } return []activityshared.Event{appServerCompactionNoticeEvent(session, turnID, messageID, status)} } switch itemType { case "agentMessage": if !completed { return nil } normalizer.ApplyAssistantFinalText(normalizeAppServerFileCitations(asStringRaw(item["text"]))) return normalizer.Finish(session, turnID, messageStreamStateCompleted) case "plan": if !completed { return nil } // Render the proposed plan as a dedicated card instead of merging it // into the assistant bubble: close any streaming text first, then // emit a standalone message tagged messageKind=plan for the GUI. events := normalizer.Finish(session, turnID, messageStreamStateCompleted) planMessageID := "plan:" + firstNonEmpty(asString(item["id"]), newID()) events = append(events, newTurnActivityEventWithID( session, planMessageID, EventMessage, turnID, messageStreamStateCompleted, RoleAssistant, asStringRaw(item["text"]), map[string]any{ "messageId": planMessageID, "contentMode": messageContentModeSnapshot, "streamState": messageStreamStateCompleted, "messageKind": "plan", }, )) return events case "reasoning": if !completed { return nil } // Surface review/inline reasoning as a finalized thinking row. The // normalizer dedupes against any reasoning that already streamed as // textDelta chunks, so this is safe for both delivery modes. return normalizer.FinalizeThinkingItem(session, turnID, appServerReasoningText(item)) case "userMessage", "hookPrompt": return nil default: update, ok := appServerItemToolCallUpdate(item, completed) if !ok { return nil } events, _ := normalizer.ToolCallEvents(session, turnID, update) return events } } // appServerItemToolCallUpdate converts an app-server thread item into the // ACP-style tool_call update shape consumed by the shared normalizer. func appServerItemToolCallUpdate(item map[string]any, completed bool) (map[string]any, bool) { itemID := asString(item["id"]) status := asString(item["status"]) if status == "" { if completed { status = "completed" } else { status = "in_progress" } } update := map[string]any{ "sessionUpdate": "tool_call", "toolCallId": itemID, "status": appServerItemStatus(status), } if completed { update["sessionUpdate"] = "tool_call_update" } switch asString(item["type"]) { case "commandExecution": command := asStringRaw(item["command"]) update["title"] = firstNonEmpty(command, "Run command") update["kind"] = "execute" update["rawInput"] = map[string]any{ "command": command, "cwd": asString(item["cwd"]), } if completed { output := map[string]any{} if stdout := asStringRaw(item["aggregatedOutput"]); stdout != "" { output["stdout"] = stdout } if exitCode, ok := acpIntFromValue(item["exitCode"]); ok { output["exitCode"] = exitCode } if len(output) > 0 { update["rawOutput"] = output } } case "fileChange": update["title"] = "Edit" update["kind"] = "edit" changes, _ := item["changes"].([]any) locations := make([]any, 0, len(changes)) for _, change := range changes { if path := asString(payloadObject(change)["path"]); path != "" { locations = append(locations, map[string]any{"path": path}) } } if len(locations) > 0 { update["locations"] = locations } rawInput := map[string]any{"changes": item["changes"]} if cwd := asString(item["cwd"]); cwd != "" { rawInput["cwd"] = cwd } update["rawInput"] = rawInput case "mcpToolCall": server := asString(item["server"]) tool := asString(item["tool"]) update["title"] = strings.TrimPrefix(server+"."+tool, ".") update["kind"] = "other" if arguments := item["arguments"]; arguments != nil { update["rawInput"] = map[string]any{"arguments": arguments} } if completed { output := map[string]any{} if result := payloadObject(item["result"]); len(result) > 0 { if content := result["content"]; content != nil { update["content"] = clonePayloadValue(content) } if structuredContent := result["structuredContent"]; structuredContent != nil { output["structuredContent"] = clonePayloadValue(structuredContent) } if isError, ok := result["isError"].(bool); ok { output["isError"] = isError if isError { update["status"] = messageStreamStateFailed } } } else if resultText := asStringRaw(item["result"]); resultText != "" { output["text"] = resultText } if errText := asStringRaw(item["error"]); errText != "" { output["text"] = errText update["status"] = messageStreamStateFailed } if len(output) > 0 { update["rawOutput"] = output } } case "webSearch": // Per the Codex app-server schema, a webSearch item is {id, query, action?} // and for a `search` action the real query lives in action.query or the // action.queries[] array; the top-level `query` is often empty. Read the // action first and fall back to the top-level field so the query is not // silently dropped (which previously rendered an empty web-search row). action := payloadObject(item["action"]) queries := appServerSearchQueries(action["queries"]) query := firstNonEmpty( firstAppServerQuery(queries), asString(action["query"]), asString(item["query"]), ) url := asString(action["url"]) actionType := firstNonEmpty(asString(action["type"]), "search") rawInput := map[string]any{} if query != "" { rawInput["query"] = query } if len(queries) > 0 { rawInput["search_query"] = queries } actionOut := map[string]any{"type": actionType} if query != "" { actionOut["query"] = query } if len(queries) > 0 { actionOut["queries"] = queries } if url != "" { actionOut["url"] = url } rawInput["action"] = actionOut switch { case query != "": update["title"] = "Searching for: " + query case url != "": update["title"] = "Visiting: " + url default: update["title"] = "WebSearch" } update["kind"] = "fetch" update["rawInput"] = rawInput if query == "" && url == "" && len(queries) == 0 { // Confirmed via exported logs: the Codex app-server streams web-search // items as {query:"", action:{type:"other"}} for some models (e.g. // gpt-5.x) — it never sends the query or results, so there is nothing // for the daemon or GUI to render. Keep this at debug level (it can fire // once per search) to aid future diagnosis without flooding info logs. slog.Debug( "agent session app-server web search has no query/results from provider", "itemId", itemID, "status", status, "rawItem", appServerItemJSON(item), ) } case "dynamicToolCall": update["title"] = firstNonEmpty(asString(item["tool"]), "Tool") update["kind"] = "other" if arguments := item["arguments"]; arguments != nil { update["rawInput"] = map[string]any{"arguments": arguments} } if success, ok := item["success"].(bool); ok && completed && !success { update["status"] = messageStreamStateFailed } case "collabAgentToolCall": tool := firstNonEmpty(asString(item["tool"]), "agent") update["title"] = tool if appServerAgentControlToolName(tool) != "" { update["kind"] = "other" } else { update["kind"] = "execute" } rawInput := map[string]any{ "task": asStringRaw(item["prompt"]), "agentName": tool, } // Preserve provider receiver ids as tool output detail. Canonical child // attachment uses the child session's parentToolCallId relation. if receivers := appServerReceiverThreadIDs(item["receiverThreadIds"]); len(receivers) > 0 { ids := make([]any, 0, len(receivers)) for _, id := range receivers { ids = append(ids, id) } rawInput["receiverThreadIds"] = ids } update["rawInput"] = rawInput if completed { output := appServerCollabAgentRawOutput(item) if len(output) > 0 { update["rawOutput"] = output } else if update["status"] == messageStreamStateFailed { slog.Debug( "agent session app-server collab agent failed without output", "itemId", itemID, "tool", tool, "status", status, "rawItem", appServerItemJSON(item), ) } } case "subAgentActivity": if asString(item["kind"]) != "started" { return nil, false } childThreadID := strings.TrimSpace(asString(item["agentThreadId"])) if childThreadID == "" { return nil, false } update["sessionUpdate"] = "tool_call_update" update["status"] = messageStreamStateCompleted update["title"] = "spawnAgent" update["kind"] = "execute" update["rawInput"] = map[string]any{ "agentName": "spawnAgent", "receiverThreadIds": []any{childThreadID}, } case "imageGeneration": update["title"] = "Generate image" update["kind"] = "other" if completed { output := map[string]any{} content := make([]any, 0, 2) if revisedPrompt := strings.TrimSpace(asStringRaw(item["revisedPrompt"])); revisedPrompt != "" { if !strings.HasPrefix(strings.ToLower(revisedPrompt), "revised prompt:") { revisedPrompt = "Revised prompt: " + revisedPrompt } content = append(content, map[string]any{ "type": "content", "content": map[string]any{ "type": "text", "text": revisedPrompt, }, }) } if savedPath := asString(item["savedPath"]); savedPath != "" { output["savedPath"] = savedPath content = append(content, map[string]any{ "type": "content", "content": map[string]any{ "type": "image", "uri": savedPath, "mimeType": appServerImageMimeType(savedPath), }, }) } if len(output) > 0 { update["rawOutput"] = output } if len(content) > 0 { update["content"] = content } } case "imageView": update["title"] = "View image" update["kind"] = "read" if path := asString(item["path"]); path != "" { update["locations"] = []any{map[string]any{"path": path}} } default: return nil, false } return update, true } func appServerImageMimeType(path string) string { mimeType := strings.TrimSpace(mime.TypeByExtension(filepath.Ext(path))) if separator := strings.IndexByte(mimeType, ';'); separator >= 0 { mimeType = strings.TrimSpace(mimeType[:separator]) } if strings.HasPrefix(strings.ToLower(mimeType), "image/") { return mimeType } return "image/png" } func appServerAgentControlToolName(tool string) string { switch normalizeAgentToolToken(tool) { case "closeagent": return "CloseAgent" case "wait": return "Wait" default: return "" } } func appServerCollabAgentRawOutput(item map[string]any) map[string]any { output := map[string]any{} if message := firstNonEmpty( appServerOutputText(item["error"]), appServerOutputText(item["message"]), ); message != "" { output["message"] = message } if result := item["result"]; result != nil { output["result"] = clonePayloadValue(result) } if text := firstNonEmpty( appServerOutputText(item["output"]), appServerOutputText(item["stdout"]), ); text != "" { output["output"] = text } if stderr := appServerOutputText(item["stderr"]); stderr != "" { output["stderr"] = stderr } return output } func appServerOutputText(value any) string { if text := asStringRaw(value); strings.TrimSpace(text) != "" { return text } record := payloadObject(value) return firstNonEmpty( asStringRaw(record["message"]), asStringRaw(record["text"]), asStringRaw(record["output"]), asStringRaw(record["result"]), ) } type appServerNotificationRoute struct { session Session child *codexAppServerThreadContext turnID string normalizer *acpTurnNormalizer events []activityshared.Event registeredChildCount int drop bool }