package acp

import (
	"context"
	"encoding/json"
	"errors"
	"path/filepath"

	"albedo/cli/internal/daemon"
)

func (a *adapter) list(message request) {
	var params struct {
		Cwd    *string `json:"cwd"`
		Cursor *string `json:"cursor"`
	}
	if decodeParams(message.Params, &params) != nil || params.Cwd != nil && !filepath.IsAbs(*params.Cwd) {
		a.respond(message.ID, nil, problem(-32602, "invalid_params", "cwd must be absolute"))
		return
	}
	if err := a.connectBackend(); err != nil {
		a.respond(message.ID, nil, err)
		return
	}
	page, err := a.backend.List(a.ctx, params.Cwd, params.Cursor)
	if err != nil {
		a.respond(message.ID, nil, err)
		return
	}
	sessions := []map[string]any{}
	for _, row := range page.Items {
		item := map[string]any{"sessionId": row.ID, "cwd": row.Workspace, "title": row.Name}
		if row.ActivityAt != nil {
			item["updatedAt"] = *row.ActivityAt
		}
		sessions = append(sessions, item)
	}
	result := map[string]any{"sessions": sessions}
	if page.Next != nil {
		result["nextCursor"] = *page.Next
	}
	a.respond(message.ID, result, nil)
}

// consumeLocked routes live events into the bounded observation buffer during
// history replay or attachment validation. The caller holds the session lock.
func (s *session) consumeLocked(event daemon.StreamEvent) error {
	if !s.buffering {
		s.observe(event)
		return nil
	}
	if s.bufferFailure != nil {
		return s.bufferFailure
	}
	if event.Type == daemon.EventReset {
		s.failBufferingLocked(problem(-32603, "observation_uncertain", "Session stream reset during load"))
		return s.bufferFailure
	}
	encoded, err := json.Marshal(event)
	if err != nil {
		s.failBufferingLocked(err)
		return err
	}
	if s.bufferedBytes+len(encoded)+1 > maxUnsent {
		s.failBufferingLocked(problem(-32603, "load_buffer_overflow", "Live events exceeded 4 MiB during load"))
		return s.bufferFailure
	}
	s.bufferedBytes += len(encoded) + 1
	s.buffered = append(s.buffered, event)
	return nil
}

func (s *session) failBufferingLocked(err error) {
	if s.bufferFailure == nil {
		s.bufferFailure = err
	}
	if s.replayCancel != nil {
		s.replayCancel()
	}
}

func (s *session) load(message request) (failure error) {
	ctx, cancel := context.WithCancel(s.ctx)
	s.mu.Lock()
	s.replayCancel = cancel
	if s.bufferFailure != nil {
		cancel()
	}
	s.mu.Unlock()
	defer func() {
		cancel()
		s.mu.Lock()
		if s.bufferFailure != nil {
			failure = s.bufferFailure
		}
		s.mu.Unlock()
	}()

	select {
	case <-s.ready:
	case <-ctx.Done():
		return ctx.Err()
	}
	cursor := daemon.ACPHistoryCursor{Done: s.historyCutoff == 0}
	for !cursor.Done {
		events, next, err := s.adapter.backend.History(ctx, s.id, cursor, s.historyCutoff)
		if err != nil {
			return err
		}
		if next.Position < cursor.Position || next.Position > s.historyCutoff ||
			!next.Done && (next.Token == "" || next.Token == cursor.Token) {
			return errors.New("invalid history replay progress")
		}
		if err = s.replayPage(ctx, events); err != nil {
			return err
		}

		cursor = next
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.bufferFailure != nil {
		return s.bufferFailure
	}
	s.adapter.respond(message.ID, map[string]any{"configOptions": modelOptions(s.models)}, nil)
	return s.drainBufferedLocked()
}

func (s *session) replay(event daemon.StreamEvent) error {
	switch event.Type {
	case daemon.EventUser:
		if event.Source == "chat" {
			s.chunk("user_message_chunk", event.EntryID, event.Text)
		}
	case daemon.EventMessage, daemon.EventText:
		s.chunk("agent_message_chunk", event.EntryID, event.Text)
	case daemon.EventThinking:
		s.chunk("agent_thought_chunk", event.EntryID, event.Text)
	case daemon.EventTool:
		if event.ProgressCallID == "" || event.ToolStatus != "completed" && event.ToolStatus != "failed" {
			return errors.New("invalid history tool identity or outcome")
		}
		s.finishTool(event)
		// Only the captured active turn can also arrive after the live cursor.
		// Historical calls need no session-lifetime deduplication records.
		if s.turn == "" || event.TurnID != s.turn {
			delete(s.tools, event.ProgressCallID)
		}
	}
	s.limitReconciliation()
	if s.uncertain {
		return s.failure
	}
	return s.adapter.writer.failure()
}

// Drain the writer within each page because one hydrated page can exceed its
// queue limit. Live events continue to buffer while replay waits.
func (s *session) replayPage(ctx context.Context, events []daemon.StreamEvent) error {
	for _, event := range events {
		for {
			piece := event
			switch event.Type {
			case daemon.EventUser, daemon.EventMessage, daemon.EventText, daemon.EventThinking:
				piece.Text = boundedText(event.Text, 128*1024)
				event.Text = event.Text[len(piece.Text):]
			default:
				event.Text = ""
			}
			s.mu.Lock()
			err := s.bufferFailure
			if err == nil {
				err = s.replay(piece)
			}
			s.mu.Unlock()
			if err != nil {
				return err
			}
			select {
			case <-s.adapter.writer.flush():
			case <-s.adapter.writer.done:
				return errors.Join(errors.New("stdout writer stopped during load"), s.adapter.writer.failure())
			case <-ctx.Done():
				return ctx.Err()
			}
			if event.Text == "" {
				break
			}
		}
	}
	return nil
}

func (s *session) pauseObservation() {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.buffering = true
}

func (s *session) resumeObservation() {
	s.mu.Lock()
	defer s.mu.Unlock()
	if err := s.drainBufferedLocked(); err != nil {
		s.lose("observation_uncertain", err.Error())
	}
}

func (s *session) drainBufferedLocked() error {
	s.buffering = false
	defer func() {
		s.buffered = nil
		s.bufferedBytes = 0
	}()
	if s.bufferFailure != nil {
		return s.bufferFailure
	}
	for _, event := range s.buffered {
		s.observe(event)
	}
	return nil
}