// These tests hold a history request open and control worker order to reproduce
// attachment races that a subprocess E2E cannot schedule reliably.
package acp

import (
	"context"
	"encoding/json"
	"net"
	"net/http"
	"net/http/httptest"
	"testing"
	"time"

	"albedo/cli/internal/app"
	"albedo/cli/internal/daemon"
	"albedo/cli/internal/daemon/protocol"
)

type setupBackend struct {
	Backend
	conn    *daemon.Connection
	entered chan struct{}
	unblock chan struct{}
	calls   chan string
}

func (b *setupBackend) Models(context.Context) (app.ACPModels, error) {
	return app.ACPModels{Current: "fixture/model", Available: []string{"fixture/model"}}, nil
}

func (b *setupBackend) CurrentModel(context.Context, string) (string, error) {
	return "fixture/model", nil
}

func (b *setupBackend) Observer(id string, cursor protocol.Cursor) *daemon.ChatClient {
	client := daemon.NewChatClient(b.conn, id)
	client.SeedCursor(cursor)
	return client
}
func (b *setupBackend) Resume(_ context.Context, id, cwd string, _ []protocol.ACPMCPServer) (protocol.ACPAttachment, error) {
	b.calls <- cwd
	cutoff := int64(0)
	if cwd == "/blocked" {
		cutoff = 1
	}
	return protocol.ACPAttachment{SessionID: id, HistoryCutoff: cutoff}, nil
}
func (b *setupBackend) New(ctx context.Context, cwd string, servers []protocol.ACPMCPServer) (protocol.ACPAttachment, error) {
	return b.Resume(ctx, "00000000-0000-0000-0000-000000000002", cwd, servers)
}
func (b *setupBackend) Release(context.Context, string) error {
	panic("replacement released the daemon lease")
}
func (b *setupBackend) List(context.Context, *string, *string) (protocol.SessionPage, error) {
	return protocol.SessionPage{}, nil
}
func (b *setupBackend) History(ctx context.Context, _ string, _ daemon.ACPHistoryCursor, _ int64) ([]daemon.StreamEvent, daemon.ACPHistoryCursor, error) {
	close(b.entered)
	select {
	case <-b.unblock:
		return nil, daemon.ACPHistoryCursor{Position: 1, Done: true}, nil
	case <-ctx.Done():
		return nil, daemon.ACPHistoryCursor{}, ctx.Err()
	}
}
func TestBlockedLoadAllowsOtherSessionsAndOrdersReplacement(t *testing.T) {
	s, wire := testSession(t)
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "text/event-stream")
		w.WriteHeader(200)
		w.(http.Flusher).Flush()
		<-r.Context().Done()
	}))
	t.Cleanup(server.Close)
	b := &setupBackend{conn: daemon.NewConnection(daemon.ConnectionSnapshot{Port: server.Listener.Addr().(*net.TCPAddr).Port}, nil), entered: make(chan struct{}), unblock: make(chan struct{}), calls: make(chan string, 8)}
	a := s.adapter
	a.backend = b
	a.sessions = map[string]*session{}
	t.Cleanup(func() {
		a.cancel()
		a.mu.Lock()
		for _, current := range a.sessions {
			current.stop()
		}
		a.mu.Unlock()
	})
	// Cancel streams before waiting for the HTTP server cleanup.
	t.Cleanup(a.cancel)
	requestFor := func(method, id, cwd string) request {
		params, _ := json.Marshal(map[string]any{"sessionId": id, "cwd": cwd, "mcpServers": []any{}})
		return request{Method: method, ID: json.RawMessage("1"), Params: params}
	}
	id := "00000000-0000-0000-0000-000000000001"
	loadDone := make(chan struct{})
	go func() { a.setupRequest(requestFor("session/load", id, "/blocked"))(); close(loadDone) }()
	select {
	case <-b.entered:
	case <-time.After(3 * time.Second):
		t.Fatal("history not reached")
	}
	// Reserve replacements in request order, but start the later worker first.
	first := a.setupRequest(requestFor("session/resume", id, "/first"))
	second := a.setupRequest(requestFor("session/resume", id, "/second"))
	done := make(chan struct{})
	go func() { second(); close(done) }()
	go first()
	otherDone := make(chan struct{})
	go func() {
		a.list(request{ID: json.RawMessage("2"), Params: json.RawMessage("{}")})
		a.setupRequest(requestFor("session/new", "", "/new"))()
		a.setupRequest(requestFor("session/resume", "00000000-0000-0000-0000-000000000003", "/other"))()
		close(otherDone)
	}()
	select {
	case <-otherDone:
	case <-time.After(3 * time.Second):
		t.Fatal("blocked load stalled another session")
	}
	close(b.unblock)
	select {
	case <-loadDone:
	case <-time.After(3 * time.Second):
		t.Fatal("load did not finish")
	}
	select {
	case <-done:
	case <-time.After(3 * time.Second):
		t.Fatal("replacement did not finish")
	}
	var calls []string
	for len(b.calls) > 0 {
		calls = append(calls, <-b.calls)
	}
	if len(calls) != 5 || calls[0] != "/blocked" || calls[3] != "/first" || calls[4] != "/second" {
		t.Fatalf("attachment order %v", calls)
	}
	a.mu.Lock()
	current := a.sessions[id]
	a.mu.Unlock()
	if current == nil || current.ctx.Err() != nil {
		t.Fatal("stale replacement installed")
	}
	// Each request must produce a response, including the independent list.
	<-a.writer.flush()
	if len(wire.written) != 6 {
		t.Fatalf("responses: %d", len(wire.written))
	}
}

// The controlled rejection occurs after a live event crosses the old observer.
// A subprocess cannot force that capture/replacement ordering reliably.
type refusedResumeBackend struct {
	Backend
	prior *session
}

func (b refusedResumeBackend) Models(context.Context) (app.ACPModels, error) {
	return app.ACPModels{}, nil
}

func (b refusedResumeBackend) Resume(context.Context, string, string, []protocol.ACPMCPServer) (protocol.ACPAttachment, error) {
	b.prior.mu.Lock()
	defer b.prior.mu.Unlock()
	if err := b.prior.consumeLocked(daemon.StreamEvent{Type: daemon.EventText, TurnID: "turn", MessageID: "live", Text: "during validation"}); err != nil {
		return protocol.ACPAttachment{}, err
	}
	return protocol.ACPAttachment{}, &daemon.APIError{StatusCode: 409, Code: "workspace_mismatch"}
}
func (b refusedResumeBackend) Release(context.Context, string) error {
	panic("a rejected replacement released its lease")
}
func TestRejectedReplacementKeepsObserverAndBufferedOutput(t *testing.T) {
	s, wire := testSession(t)
	s.id = "00000000-0000-0000-0000-000000000001"
	s.turn = "turn"
	s.adapter.sessions = map[string]*session{s.id: s}
	s.adapter.backend = refusedResumeBackend{prior: s}
	s.adapter.setup(request{Method: "session/resume", ID: json.RawMessage("1"), Params: json.RawMessage(`{"sessionId":"00000000-0000-0000-0000-000000000001","cwd":"/wrong","mcpServers":[]}`)})
	<-s.adapter.writer.flush()
	if s.adapter.sessions[s.id] != s || s.ctx.Err() != nil || s.uncertain || s.buffering || s.live["live"] == nil || s.live["live"].bytes != len("during validation") {
		t.Fatal("rejected replacement changed the existing observer")
	}
	if len(wire.written) != 2 {
		t.Fatalf("expected buffered update and refusal, got %d messages", len(wire.written))
	}
}