fix(tools): parse Gemini and Codex streaming JSON formats

parseStreamLine only handled Claude's format. Now handles:
- Gemini: {"type":"message","role":"assistant","content":"..."}
- Codex: {"type":"item.completed","item":{"type":"agent_message","text":"..."}}

Also adds streaming integration tests for Gemini and Codex,
gated on PATH detection and auth availability.

Author: priyanshujain

agent/tools/agent_stream.go

@@ -128,25 +128,42 @@ func (r *StreamRunner) buildEnv() []string {
 	return env
 }
 
-// streamJSON is the minimal structure of a Claude stream-json event.
+// streamJSON is a unified structure for parsing streaming events from all CLIs.
 type streamJSON struct {
 	Type    string `json:"type"`
 	Content string `json:"content"`
 	Result  string `json:"result"`
+	Role    string `json:"role"`
+	Delta   bool   `json:"delta"`
+	Item    struct {
+		Type string `json:"type"`
+		Text string `json:"text"`
+	} `json:"item"`
 }
 
 func parseStreamLine(line []byte) StreamEvent {
 	var sj streamJSON
 	if err := json.Unmarshal(line, &sj); err != nil {
 		return StreamEvent{}
 	}
+	if sj.Type == "" {
+		return StreamEvent{}
+	}
+
+	// Gemini: {"type":"message","role":"assistant","content":"Hello","delta":true}
+	if sj.Type == "message" && sj.Role == "assistant" {
+		return StreamEvent{Type: "text", Content: sj.Content}
+	}
+
+	// Codex: {"type":"item.completed","item":{"type":"agent_message","text":"Hello"}}
+	if sj.Type == "item.completed" && sj.Item.Type == "agent_message" {
+		return StreamEvent{Type: "text", Content: sj.Item.Text}
+	}
+
+	// Claude/Gemini: {"type":"text"|"result", "content":"...", "result":"..."}
 	content := sj.Content
 	if content == "" {
 		content = sj.Result
 	}
-	typ := sj.Type
-	if typ == "" {
-		return StreamEvent{}
-	}
-	return StreamEvent{Type: typ, Content: content}
+	return StreamEvent{Type: sj.Type, Content: content}
 }

agent/tools/agent_stream_test.go

@@ -3,6 +3,7 @@ package tools
 import (
 	"context"
 	"os/exec"
+	"strings"
 	"testing"
 	"time"
 )
@@ -45,6 +46,52 @@ func TestParseStreamLine_InvalidJSON(t *testing.T) {
 	}
 }
 
+func TestParseStreamLine_GeminiMessage(t *testing.T) {
+	line := []byte(`{"type":"message","role":"assistant","content":"Hello","delta":true}`)
+	evt := parseStreamLine(line)
+	if evt.Type != "text" {
+		t.Errorf("Type = %q, want text", evt.Type)
+	}
+	if evt.Content != "Hello" {
+		t.Errorf("Content = %q", evt.Content)
+	}
+}
+
+func TestParseStreamLine_GeminiResult(t *testing.T) {
+	line := []byte(`{"type":"result","status":"success","stats":{"total_tokens":100}}`)
+	evt := parseStreamLine(line)
+	if evt.Type != "result" {
+		t.Errorf("Type = %q, want result", evt.Type)
+	}
+}
+
+func TestParseStreamLine_GeminiUserMessage(t *testing.T) {
+	line := []byte(`{"type":"message","role":"user","content":"prompt"}`)
+	evt := parseStreamLine(line)
+	if evt.Type == "text" {
+		t.Error("user messages should not be parsed as text events")
+	}
+}
+
+func TestParseStreamLine_CodexItemCompleted(t *testing.T) {
+	line := []byte(`{"type":"item.completed","item":{"id":"item_1","type":"agent_message","text":"Hello"}}`)
+	evt := parseStreamLine(line)
+	if evt.Type != "text" {
+		t.Errorf("Type = %q, want text", evt.Type)
+	}
+	if evt.Content != "Hello" {
+		t.Errorf("Content = %q", evt.Content)
+	}
+}
+
+func TestParseStreamLine_CodexReasoningIgnored(t *testing.T) {
+	line := []byte(`{"type":"item.completed","item":{"type":"reasoning","text":"thinking..."}}`)
+	evt := parseStreamLine(line)
+	if evt.Type == "text" {
+		t.Error("reasoning items should not be parsed as text events")
+	}
+}
+
 func TestParseStreamLine_EmptyType(t *testing.T) {
 	line := []byte(`{"content":"orphan"}`)
 	evt := parseStreamLine(line)
@@ -168,3 +215,59 @@ func TestStreamRunner_RealClaude(t *testing.T) {
 		t.Error("expected at least one event")
 	}
 }
+
+func TestStreamRunner_RealGemini(t *testing.T) {
+	if _, err := exec.LookPath("gemini"); err != nil {
+		t.Skip("gemini not on PATH")
+	}
+	agents := DetectAgents()
+	var info AgentInfo
+	for _, a := range agents {
+		if a.Kind == AgentGemini {
+			info = a
+			break
+		}
+	}
+	r := NewStreamRunner(info)
+	var eventCount int
+	out, err := r.RunStream(context.Background(), "Say hello in exactly one word.", 30*time.Second, func(evt StreamEvent) {
+		eventCount++
+	})
+	if err != nil {
+		if strings.Contains(err.Error(), "Permission") || strings.Contains(err.Error(), "denied") || strings.Contains(err.Error(), "auth") {
+			t.Skipf("gemini auth not configured: %v", err)
+		}
+		t.Fatalf("RunStream: %v", err)
+	}
+	if out == "" {
+		t.Error("expected non-empty output")
+	}
+}
+
+func TestStreamRunner_RealCodex(t *testing.T) {
+	if _, err := exec.LookPath("codex"); err != nil {
+		t.Skip("codex not on PATH")
+	}
+	agents := DetectAgents()
+	var info AgentInfo
+	for _, a := range agents {
+		if a.Kind == AgentCodex {
+			info = a
+			break
+		}
+	}
+	r := NewStreamRunner(info)
+	var eventCount int
+	out, err := r.RunStream(context.Background(), "Say hello in exactly one word.", 30*time.Second, func(evt StreamEvent) {
+		eventCount++
+	})
+	if err != nil {
+		if strings.Contains(err.Error(), "auth") || strings.Contains(err.Error(), "API key") || strings.Contains(err.Error(), "login") {
+			t.Skipf("codex auth not configured: %v", err)
+		}
+		t.Fatalf("RunStream: %v", err)
+	}
+	if out == "" {
+		t.Error("expected non-empty output")
+	}
+}