// A daemon cannot emit deliberately malformed wire data. Controlled HTTP peers
// exercise batch atomicity and failure classification through the public stream API.
package daemon

import (
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net"
	"net/http"
	"net/http/httptest"
	"os"
	"strings"
	"sync/atomic"
	"testing"
)

func TestStreamRejectsMalformedBatchesWithoutLosingTheCursor(t *testing.T) {
	data, err := os.ReadFile("../../../test/fixtures/session_stream.json")
	if err != nil {
		t.Fatal(err)
	}
	var fixture struct {
		Malformed map[string]string `json:"malformed_batches"`
	}
	if err := json.Unmarshal(data, &fixture); err != nil {
		t.Fatal(err)
	}
	if len(fixture.Malformed) == 0 {
		t.Fatal("empty stream contract fixture")
	}
	for name, batch := range fixture.Malformed {
		t.Run(name, func(t *testing.T) {
			var requests atomic.Int32
			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				w.Header().Set("Content-Type", "text/event-stream")
				switch requests.Add(1) {
				case 1:
					_, _ = io.WriteString(w, "data: "+`{"generation":"generation-a","cursor":7,"events":[{"type":"reset"},{"type":"text","text":"accepted"}]}`+"\n\n")
				case 2:
					_, _ = fmt.Fprintf(w, "data: %s\n\n", batch)
				default:
					if got := r.URL.Query().Get("after_seq"); got != "7" {
						t.Errorf("failed batch lost committed cursor: %s", got)
					}
					_, _ = io.WriteString(w, "data: "+`{"generation":"generation-a","cursor":8,"events":[{"type":"text","text":"replayed"}]}`+"\n\n")
				}
			}))
			defer server.Close()
			client := NewChatClient(NewConnection(ConnectionSnapshot{Port: server.Listener.Addr().(*net.TCPAddr).Port}, nil), "session")
			var delivered []string
			consume := func(event StreamEvent) error {
				if event.Type == EventReset {
					return nil
				}
				if event.Type != EventText {
					t.Errorf("unexpected event delivered: %+v", event)
				}
				delivered = append(delivered, event.Text)
				return nil
			}
			if err := client.Stream(t.Context(), 0, consume); err != nil {
				t.Fatal(err)
			}
			consumed := 0
			err := client.StreamWithProgress(t.Context(), 0, consume, func() { consumed++ })
			if consumed != 0 {
				t.Fatal("malformed batch reported consumption")
			}
			failure, ok := errors.AsType[*StreamError](err)
			if !ok || failure.Kind != StreamProtocol {
				t.Fatalf("malformed batch was not a protocol failure: %v", err)
			}
			if len(delivered) != 1 || delivered[0] != "accepted" {
				t.Fatalf("malformed batch partially delivered: %v", delivered)
			}
			if err := client.Stream(t.Context(), 0, consume); err != nil {
				t.Fatal(err)
			}
			if len(delivered) != 2 || delivered[1] != "replayed" || requests.Load() != 3 {
				t.Fatalf("replay lost events or retried internally: %v, requests=%d", delivered, requests.Load())
			}
		})
	}
}

func TestStreamIgnoresUnknownKindsAndCommitsTheirCursor(t *testing.T) {
	var requests atomic.Int32
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "text/event-stream")
		if requests.Add(1) == 1 {
			_, _ = io.WriteString(w, "data: "+`{"generation":"generation-a","cursor":12,"events":[{"type":"reset"},{"type":"future_output","text":"not assistant output","args":false}]}`+"\n\n")
		} else if got := r.URL.Query().Get("after_seq"); got != "12" {
			t.Errorf("ignored additive batch lost cursor: %s", got)
		}
	}))
	defer server.Close()
	client := NewChatClient(NewConnection(ConnectionSnapshot{Port: server.Listener.Addr().(*net.TCPAddr).Port}, nil), "session")
	consumed := 0
	for range 2 {
		if err := client.StreamWithProgress(t.Context(), 0, func(event StreamEvent) error {
			if event.Type != EventReset {
				t.Errorf("unknown event delivered: %+v", event)
			}
			return nil
		}, func() { consumed++ }); err != nil {
			t.Fatal(err)
		}
	}
	if consumed != 1 {
		t.Fatalf("ignored-only batch did not report consumption: %d", consumed)
	}
}

func TestStreamCallbackFailureRetainsCauseAndUncommittedCursor(t *testing.T) {
	var requests atomic.Int32
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "text/event-stream")
		n := requests.Add(1)
		if n == 1 {
			_, _ = io.WriteString(w, "data: "+`{"generation":"generation-a","cursor":7,"events":[{"type":"reset"}]}`+"\n\n")
			return
		}
		if n == 3 && (r.URL.Query().Get("after_seq") != "7" || r.URL.Query().Get("after_generation") != "generation-a") {
			t.Error("callback failure committed the batch cursor")
		}
		_, _ = io.WriteString(w, "data: "+`{"generation":"generation-b","cursor":8,"events":[{"type":"reset"},{"type":"text","text":"first"},{"type":"text","text":"second"}]}`+"\n\n")
	}))
	defer server.Close()
	client := NewChatClient(NewConnection(ConnectionSnapshot{Port: server.Listener.Addr().(*net.TCPAddr).Port}, nil), "session")
	if err := client.Stream(t.Context(), 0, func(StreamEvent) error { return nil }); err != nil {
		t.Fatal(err)
	}
	cause := errors.New("consumer stopped")
	var callbacks int
	err := client.Stream(t.Context(), 0, func(event StreamEvent) error {
		if event.Type == EventReset {
			return nil
		}
		callbacks++
		if callbacks == 2 {
			return cause
		}
		return nil
	})
	failure, ok := errors.AsType[*StreamError](err)
	if !ok || failure.Kind != StreamTerminal || !errors.Is(err, cause) {
		t.Fatalf("callback failure lost classification or cause: %v", err)
	}
	if requests.Load() != 2 || callbacks != 2 {
		t.Fatalf("callback failure changed delivery or retried: requests=%d, callbacks=%d", requests.Load(), callbacks)
	}
	if err := client.Stream(t.Context(), 0, func(StreamEvent) error { return nil }); err != nil {
		t.Fatal(err)
	}
}

func TestStreamClassifiesHTTPRefusalsWithoutRetryingThem(t *testing.T) {
	for _, test := range []struct {
		status int
		kind   StreamFailureKind
	}{
		{http.StatusNotFound, StreamTerminal}, {http.StatusForbidden, StreamTerminal},
		{http.StatusBadRequest, StreamTerminal}, {http.StatusRequestTimeout, StreamTransient},
		{http.StatusTooManyRequests, StreamTransient}, {http.StatusServiceUnavailable, StreamTransient},
	} {
		t.Run(http.StatusText(test.status), func(t *testing.T) {
			var requests atomic.Int32
			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				requests.Add(1)
				w.WriteHeader(test.status)
				_, _ = io.WriteString(w, `{"error":"stream unavailable"}`)
			}))
			defer server.Close()
			client := NewChatClient(NewConnection(ConnectionSnapshot{Port: server.Listener.Addr().(*net.TCPAddr).Port}, nil), "session")
			err := client.Stream(t.Context(), 0, func(StreamEvent) error {
				t.Error("HTTP refusal delivered an event")
				return nil
			})
			failure, ok := errors.AsType[*StreamError](err)
			api, apiOK := errors.AsType[*APIError](err)
			if !ok || failure.Kind != test.kind || !apiOK || api.StatusCode != test.status || requests.Load() != 1 {
				t.Fatalf("wrong refusal classification, cause, or retry: %v, requests=%d", err, requests.Load())
			}
		})
	}
}

func TestStreamRejectsNonSSESuccessResponses(t *testing.T) {
	for _, media := range []string{"", "text/html", "application/json"} {
		t.Run(media, func(t *testing.T) {
			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				w.Header()["Content-Type"] = nil
				if media != "" {
					w.Header().Set("Content-Type", media)
				}
				_, _ = io.WriteString(w, `{"error":"this is not a stream"}`)
			}))
			defer server.Close()
			client := NewChatClient(NewConnection(ConnectionSnapshot{Port: server.Listener.Addr().(*net.TCPAddr).Port}, nil), "session")
			err := client.Stream(t.Context(), 0, func(StreamEvent) error {
				t.Error("non-SSE response delivered an event")
				return nil
			})
			failure, ok := errors.AsType[*StreamError](err)
			if !ok || failure.Kind != StreamProtocol {
				t.Fatalf("wrong media type treated as recoverable EOF: %v", err)
			}
		})
	}
}

func TestStreamRecoveryRequiresResetBeforeDeliveringHistory(t *testing.T) {
	var requests atomic.Int32
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "text/event-stream")
		switch requests.Add(1) {
		case 1:
			_, _ = io.WriteString(w, "data: "+`{"generation":"generation-a","cursor":20,"events":[{"type":"reset"}]}`+"\n\n")
		case 2:
			_, _ = io.WriteString(w, "data: "+`{"generation":"generation-a","cursor":1,"events":[{"type":"text","text":"not durable reset"}]}`+"\n\n")
		case 3:
			_, _ = io.WriteString(w, "data: "+`{"generation":"generation-a","cursor":1,"events":[{"type":"reset"},{"type":"message","role":"assistant","text":"durable"}]}`+"\n\n")
		default:
			if r.URL.Query().Get("after_seq") != "1" {
				t.Error("validated reset failed to commit its cursor")
			}
		}
	}))
	defer server.Close()
	client := NewChatClient(NewConnection(ConnectionSnapshot{Port: server.Listener.Addr().(*net.TCPAddr).Port}, nil), "session")
	consume := func(StreamEvent) error { return nil }
	if err := client.Stream(t.Context(), 0, consume); err != nil {
		t.Fatal(err)
	}
	client.ResetStream()
	err := client.Stream(t.Context(), 0, func(StreamEvent) error {
		t.Error("non-reset recovery data escaped")
		return nil
	})
	failure, ok := errors.AsType[*StreamError](err)
	if !ok || failure.Kind != StreamProtocol {
		t.Fatalf("missing reset accepted: %v", err)
	}
	var replayed bool
	if err := client.Stream(t.Context(), 0, func(event StreamEvent) error {
		if event.Type == EventMessage {
			replayed = event.Replayed
		}
		return nil
	}); err != nil {
		t.Fatal(err)
	}
	if !replayed {
		t.Fatal("reset snapshot was not marked as replayed history")
	}
	if err := client.Stream(t.Context(), 0, consume); err != nil {
		t.Fatal(err)
	}
}

func TestStreamClassifiesExplicitSSEFailures(t *testing.T) {
	for _, test := range []struct {
		name, payload string
		kind          StreamFailureKind
	}{
		{"reported failure", `{"error":"session removed"}`, StreamTerminal},
		{"malformed JSON", `{"error":`, StreamProtocol},
		{"malformed field", `{"error":false}`, StreamProtocol},
		{"missing reason", `{}`, StreamProtocol},
	} {
		t.Run(test.name, func(t *testing.T) {
			var requests atomic.Int32
			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				requests.Add(1)
				w.Header().Set("Content-Type", "text/event-stream")
				_, _ = fmt.Fprintf(w, "event: error\ndata: %s\n\n", test.payload)
			}))
			defer server.Close()
			client := NewChatClient(NewConnection(ConnectionSnapshot{Port: server.Listener.Addr().(*net.TCPAddr).Port}, nil), "session")
			err := client.Stream(t.Context(), 0, func(StreamEvent) error {
				t.Error("SSE failure became transcript output")
				return nil
			})
			failure, ok := errors.AsType[*StreamError](err)
			if !ok || failure.Kind != test.kind || requests.Load() != 1 {
				t.Fatalf("wrong explicit failure or automatic retry: %v, requests=%d", err, requests.Load())
			}
		})
	}
}

func TestStreamGenerationResetReplacesUnfinishedArguments(t *testing.T) {
	var requests atomic.Int32
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "text/event-stream")
		switch requests.Add(1) {
		case 1:
			if r.URL.Query().Has("after_seq") || r.URL.Query().Has("after_generation") {
				t.Error("initial subscription sent a cursor")
			}
			_, _ = io.WriteString(w, "data: "+`{"generation":"old","cursor":10,"events":[{"type":"reset"},{"type":"arguments_delta","callId":"same","name":"python","text":"{\"code\":\"old"}]}`+"\n\n")
		case 2:
			_, _ = io.WriteString(w, "data: "+`{"generation":"new","cursor":11,"events":[{"type":"text","text":"must stay hidden"}]}`+"\n\n")
		case 3:
			if r.URL.Query().Get("after_seq") != "10" || r.URL.Query().Get("after_generation") != "old" {
				t.Error("invalid generation change replaced the consumed pair")
			}
			_, _ = io.WriteString(w, "data: "+`{"generation":"new","cursor":1,"events":[{"type":"reset"},{"type":"arguments_delta","callId":"same","name":"python","text":"{\"code\":\"new()\"}"}]}`+"\n\n")
		default:
			if r.URL.Query().Get("after_seq") != "1" || r.URL.Query().Get("after_generation") != "new" {
				t.Error("successful reset failed to replace both cursor values")
			}
		}
	}))
	defer server.Close()
	client := NewChatClient(NewConnection(ConnectionSnapshot{Port: server.Listener.Addr().(*net.TCPAddr).Port}, nil), "session")
	var previews []string
	consume := func(event StreamEvent) error {
		if event.Type == EventText {
			t.Error("invalid generation events escaped")
		}
		if event.Progress != nil && event.Progress.Code != nil {
			previews = append(previews, event.Progress.Code.Text)
		}
		return nil
	}
	if err := client.Stream(t.Context(), 0, consume); err != nil {
		t.Fatal(err)
	}
	err := client.Stream(t.Context(), 0, consume)
	failure, ok := errors.AsType[*StreamError](err)
	if !ok || failure.Kind != StreamProtocol {
		t.Fatalf("generation change without reset accepted: %v", err)
	}
	for range 2 {
		if err := client.Stream(t.Context(), 0, consume); err != nil {
			t.Fatal(err)
		}
	}
	if len(previews) != 2 || previews[0] != "old" || previews[1] != "new()" {
		t.Fatalf("reset mixed old and new arguments: %q", previews)
	}
}

func TestInitialStreamRequiresLeadingReset(t *testing.T) {
	for _, events := range []string{`[]`, `[{"type":"text","text":"must stay hidden"}]`} {
		t.Run(events, func(t *testing.T) {
			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				w.Header().Set("Content-Type", "text/event-stream")
				_, _ = fmt.Fprintf(w, "data: {\"generation\":\"new\",\"cursor\":0,\"events\":%s}\n\n", events)
			}))
			defer server.Close()
			client := NewChatClient(NewConnection(ConnectionSnapshot{Port: server.Listener.Addr().(*net.TCPAddr).Port}, nil), "session")
			err := client.Stream(t.Context(), 0, func(StreamEvent) error { t.Error("initial events escaped without reset"); return nil })
			failure, ok := errors.AsType[*StreamError](err)
			if !ok || failure.Kind != StreamProtocol {
				t.Fatalf("initial batch without reset accepted: %v", err)
			}
		})
	}
}

func TestStreamLoadsHistoryDespiteInvalidDisplayTraces(t *testing.T) {
	unicodeTrace, err := json.Marshal(ToolTrace{
		Activities: []ToolActivity{{Kind: "read", Target: strings.Repeat("界", 1000)}},
		Changes: []FileChange{
			{Kind: "diff", Path: strings.Repeat("界", 1000), Diff: strings.Repeat("🙂", 16000)},
			{Kind: "unavailable", Path: "binary", Reason: strings.Repeat("界", 1000)},
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	cases := []struct {
		name      string
		trace     string
		wantTrace bool
	}{
		{"missing", "", false},
		{"null", "null", false},
		{"wrong type", `"unusable"`, false},
		{"wrong nested type", `{"activities":false}`, false},
		{"unknown activity", `{"activities":[{"kind":"unknown","target":"file"}]}`, false},
		{"oversized target", `{"activities":[{"kind":"read","target":"` + strings.Repeat("界", 1001) + `"}]}`, false},
		{"oversized diff", `{"changes":[{"kind":"diff","path":"file","diff":"` + strings.Repeat("🙂", 16001) + `"}]}`, false},
		{"negative line count", `{"changes":[{"kind":"diff","path":"file","added":-1}]}`, false},
		{"unicode at limits", string(unicodeTrace), true},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			traceField := ""
			if tc.trace != "" {
				traceField = `,"trace":` + tc.trace
			}
			batch := `{"generation":"generation-a","cursor":12,"events":[{"type":"reset"},{"type":"tool","callId":"call","name":"python","args":"{}","result":"saved output"` + traceField + `},{"type":"text","text":"after tool"}]}`
			var requests atomic.Int32
			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				w.Header().Set("Content-Type", "text/event-stream")
				if requests.Add(1) == 1 {
					_, _ = fmt.Fprintf(w, "data: %s\n\n", batch)
				} else if got := r.URL.Query().Get("after_seq"); got != "12" {
					t.Errorf("history batch lost cursor: %s", got)
				}
			}))
			defer server.Close()
			client := NewChatClient(NewConnection(ConnectionSnapshot{Port: server.Listener.Addr().(*net.TCPAddr).Port}, nil), "session")
			var events []StreamEvent
			consumed := 0
			for range 2 {
				if err := client.StreamWithProgress(t.Context(), 0, func(event StreamEvent) error {
					events = append(events, event)
					return nil
				}, func() { consumed++ }); err != nil {
					t.Fatal(err)
				}
			}
			if consumed != 1 || len(events) != 3 || events[1].Type != EventTool || events[1].ToolResult != "saved output" || events[2].Text != "after tool" {
				t.Fatalf("history did not load completely: consumed=%d events=%+v", consumed, events)
			}
			if got := events[1].ToolTrace; (got != nil) != tc.wantTrace {
				t.Fatalf("unexpected display trace: %+v", got)
			} else if tc.wantTrace {
				gotJSON, err := json.Marshal(got)
				if err != nil || string(gotJSON) != string(unicodeTrace) {
					t.Fatalf("unicode trace changed: %s, %v", gotJSON, err)
				}
			}
		})
	}
}