// These tests control acknowledgement, cancellation, reset, and replacement
// ordering that a subprocess E2E cannot reproduce reliably. They also drain
// output promptly to test reconciliation limits without relying on pipe size.
package acp

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"strings"
	"testing"
	"time"

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

type fakeBackend struct {
	written chan []byte
	Backend
	submitted chan string
	ack       chan struct{}
	cancelled chan string
	receipt   protocol.InputCancellation
}

func (f *fakeBackend) Prepare(id string, blocks []json.RawMessage) (*daemon.OperationHandle, error) {
	return daemon.NewACPInput(id, "token", blocks)
}
func (f *fakeBackend) Submit(ctx context.Context, handle *daemon.OperationHandle) error {
	f.submitted <- handle.ID()
	select {
	case <-f.ack:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}
func (f *fakeBackend) Cancel(_ context.Context, _ string, id string) (protocol.InputCancellation, error) {
	f.cancelled <- id
	return f.receipt, nil
}
func testSession(t *testing.T) (*session, *fakeBackend) {
	t.Helper()
	ctx, cancel := context.WithCancel(context.Background())
	t.Cleanup(cancel)
	backend := &fakeBackend{written: make(chan []byte, 64), submitted: make(chan string, 8), ack: make(chan struct{}), cancelled: make(chan string, 8), receipt: protocol.InputCancellation{Result: "cancelled"}}
	a := &adapter{ctx: ctx, cancel: cancel, backend: backend, writer: newWriter(capturedWriter{backend.written})}
	ready := make(chan struct{})
	close(ready)
	s := &session{adapter: a, ctx: ctx, stop: cancel, id: "session", ready: ready, live: map[string]*streamedPrefix{}, boundaries: map[string]int64{}, durable: map[string]bool{}, tools: map[string]toolState{}}
	go a.writer.run(ctx)
	return s, backend
}
func receive(t *testing.T, ch <-chan string) string {
	t.Helper()
	select {
	case id := <-ch:
		return id
	case <-time.After(3 * time.Second):
		t.Fatal("operation did not arrive")
		return ""
	}
}
func TestCancelBeforeAckAndSuccessor(t *testing.T) {
	s, backend := testSession(t)
	s.prompt(request{ID: json.RawMessage("1")}, []json.RawMessage{})
	first := receive(t, backend.submitted)
	s.cancelPrompt()
	s.prompt(request{ID: json.RawMessage("2")}, []json.RawMessage{})
	select {
	case <-backend.cancelled:
		t.Fatal("cancelled before admission known")
	default:
	}
	close(backend.ack)
	if id := receive(t, backend.cancelled); id != first {
		t.Fatal("cancel identity changed")
	}
	second := receive(t, backend.submitted)
	if first == second {
		t.Fatal("successor reused identity")
	}
}
func TestCancelHeldSuccessorAndReset(t *testing.T) {
	s, backend := testSession(t)
	s.prompt(request{ID: json.RawMessage("1")}, []json.RawMessage{})
	receive(t, backend.submitted)
	s.cancelPrompt()
	s.prompt(request{ID: json.RawMessage("2")}, []json.RawMessage{})
	s.cancelPrompt()
	s.mu.Lock()
	if s.successor != nil {
		t.Fatal("held successor survived cancel")
	}
	s.observe(daemon.StreamEvent{Type: daemon.EventReset})
	if !s.uncertain {
		t.Fatal("reset did not block admission")
	}
	s.mu.Unlock()
	close(backend.ack)
}

func TestCompletionWinsCancelReceipt(t *testing.T) {
	s, backend := testSession(t)
	backend.receipt = protocol.InputCancellation{Result: "not_pending", Input: protocol.Input{Turn: &protocol.Turn{State: "completed"}}}
	s.prompt(request{ID: json.RawMessage("1")}, []json.RawMessage{})
	receive(t, backend.submitted)
	close(backend.ack)
	s.cancelPrompt()
	receive(t, backend.cancelled)
	// The receipt can run ahead of the turn's last output; the response
	// waits for the stream to finish the turn.
	select {
	case data := <-backend.written:
		t.Fatalf("answered before the stream finished the turn: %s", data)
	case <-time.After(200 * time.Millisecond):
	}
	s.mu.Lock()
	id := s.active.handle.ID()
	s.observe(daemon.StreamEvent{Type: daemon.EventTurnMembership, TurnID: "turn", SubmissionIDs: []string{id}})
	s.observe(daemon.StreamEvent{Type: daemon.EventTurnCompleted, TurnID: "turn", Source: "completed", SubmissionIDs: []string{id}})
	s.mu.Unlock()
	select {
	case data := <-backend.written:
		if !bytes.Contains(data, []byte(`"stopReason":"end_turn"`)) {
			t.Fatalf("completion did not win: %s", data)
		}
	case <-time.After(3 * time.Second):
		t.Fatal("missing completion response")
	}
}

type capturedWriter struct{ written chan []byte }

func (w capturedWriter) Write(data []byte) (int, error) {
	w.written <- bytes.Clone(data)
	return len(data), nil
}

func TestResetAtExactCursorDoesNotReplaySnapshot(t *testing.T) {
	s, _ := testSession(t)
	s.cursor = protocol.Cursor{Generation: "generation", Sequence: 42}
	snapshot := &daemon.Session{Cursor: &protocol.Cursor{Generation: "generation", Sequence: 42}}
	s.observe(daemon.StreamEvent{Type: daemon.EventReset, Snapshot: snapshot, SnapshotOrigin: true})
	if s.uncertain {
		t.Fatal("exact cursor reset lost observation")
	}
	snapshot.Cursor.Sequence++
	s.observe(daemon.StreamEvent{Type: daemon.EventReset, Snapshot: snapshot, SnapshotOrigin: true})
	if !s.uncertain {
		t.Fatal("gapped snapshot was silently accepted")
	}
}

func TestMembershipConfirmsAdmissionBeforeHTTPAck(t *testing.T) {
	s, backend := testSession(t)
	backend.receipt.Result = "interrupt_requested"
	s.prompt(request{ID: json.RawMessage("1")}, []json.RawMessage{})
	id := receive(t, backend.submitted)
	s.cancelPrompt()
	s.mu.Lock()
	s.observe(daemon.StreamEvent{Type: daemon.EventTurnMembership, TurnID: "turn", SubmissionIDs: []string{id}})
	s.mu.Unlock()
	if cancelled := receive(t, backend.cancelled); cancelled != id {
		t.Fatal("cancellation changed identity")
	}
	s.mu.Lock()
	s.observe(daemon.StreamEvent{Type: daemon.EventTurnCompleted, TurnID: "turn", Source: "interrupted", SubmissionIDs: []string{id}})
	s.mu.Unlock()
	select {
	case data := <-backend.written:
		if !bytes.Contains(data, []byte(`"stopReason":"cancelled"`)) {
			t.Fatalf("wrong stop reason: %s", data)
		}
	case <-time.After(3 * time.Second):
		t.Fatal("cancelled turn waited for HTTP acknowledgement")
	}
}

// Discarding output forces promptly consumed writes without OS pipe timing.
func TestLargePrefixAndTurnRetirement(t *testing.T) {
	s, _ := testSession(t)
	s.adapter.writer = newWriter(io.Discard)
	go s.adapter.writer.run(s.ctx)
	piece := strings.Repeat("雪", 128*1024)
	s.turn = "turn"
	for range 20 {
		s.observe(daemon.StreamEvent{Type: daemon.EventText, TurnID: "turn", MessageID: "live", Text: piece})
		<-s.adapter.writer.flush()
	}
	if s.uncertain || s.live["live"].bytes != 20*len(piece) {
		t.Fatal("large prefix lost")
	}
	s.observe(daemon.StreamEvent{Type: daemon.EventMessage, TurnID: "turn", EntryID: "durable", Text: strings.Repeat(piece, 20) + "tail", ReplacesLiveIDs: []string{"live"}})
	if s.uncertain {
		t.Fatal("matching digest rejected")
	}
	s.observe(daemon.StreamEvent{Type: daemon.EventTool, TurnID: "turn", ProgressCallID: "tool", ToolStatus: "completed"})
	s.observe(daemon.StreamEvent{Type: daemon.EventTurnCompleted, TurnID: "turn", Source: "completed"})
	if len(s.live)+len(s.tools)+len(s.durable)+len(s.boundaries)+len(s.withheld) != 0 {
		t.Fatal("finished turn retained reconciliation state")
	}
}

func TestReconciliationLimitFailsWithoutRetainingState(t *testing.T) {
	s, _ := testSession(t)
	s.adapter.writer = newWriter(io.Discard)
	go s.adapter.writer.run(s.ctx)
	s.turn = "turn"
	for i := range maxReconciliationEntries + 1 {
		s.observe(daemon.StreamEvent{Type: daemon.EventToolProgress, TurnID: "turn", Progress: &daemon.ToolProgress{CallID: fmt.Sprintf("call-%d", i), Name: "python"}})
	}
	if !s.uncertain || len(s.tools) != 0 {
		t.Fatal("identity limit did not quarantine and release state")
	}
}