package agent

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"strings"
)

// Pi's pinned RPC contract does not expose a typed invalid-session response:
// unsuccessful responses carry only success=false and error. Keep this error
// typed for command-scoped diagnostics, but never treat it as resume authority.
type piRPCResponseError struct {
	Command string
	Success bool
	Message string
}

func (e *piRPCResponseError) Error() string {
	if e == nil {
		return "pi RPC request failed"
	}
	return fmt.Sprintf("pi RPC %s: %s", strings.TrimSpace(e.Command), firstNonEmptyString(e.Message, "request failed"))
}

// requestCommand sends one RPC command and returns its correlated response.
// Callers that need a read-side snapshot (for example get_state/get_entries
// during delivery reconciliation) must inspect Data from the same response
// rather than falling back to the appendable JSONL file.
func (d *PiRPCDriver) requestCommand(ctx context.Context, command map[string]any) (piRPCFrame, error) {
	if ctx == nil {
		ctx = context.Background()
	}
	if err := ctx.Err(); err != nil {
		// No bytes can have crossed the mutation boundary before this
		// preflight check. Keep this distinguishable from a cancellation after
		// write, which is an ambiguous runtime outcome.
		return piRPCFrame{}, fmt.Errorf("%w: %w", errRuntimeInputMutationNotAttempted, err)
	}
	d.mu.Lock()
	if d.stdin == nil || d.transportClosed {
		d.mu.Unlock()
		// No bytes were attempted. This is a definite preflight failure, not
		// an ambiguous mutation outcome; callers may safely return the durable
		// owner to queued and retry it on a replacement transport.
		return piRPCFrame{}, fmt.Errorf("%w: %w", errRuntimeInputMutationNotAttempted, io.ErrClosedPipe)
	}
	d.nextID++
	id := d.nextID
	command["id"] = id
	responseCh := make(chan piRPCFrame, 1)
	d.pending[id] = responseCh
	d.mu.Unlock()

	data, err := json.Marshal(command)
	if err != nil {
		d.forgetPending(id)
		return piRPCFrame{}, err
	}
	data = append(data, '\n')
	if err := d.writeFrame(data); err != nil {
		d.forgetPending(id)
		return piRPCFrame{}, errPiRPCTransportClosed
	}
	select {
	case <-ctx.Done():
		d.forgetPending(id)
		// The frame was written before waiting for its response. A cancelled
		// wait therefore cannot tell whether Pi consumed it; preserve the
		// fail-closed mutation boundary for Process reconciliation.
		return piRPCFrame{}, errors.Join(errPiRPCTransportAmbiguous, ctx.Err())
	case response := <-responseCh:
		if response.Success != nil && !*response.Success {
			commandName, _ := command["type"].(string)
			return piRPCFrame{}, &piRPCResponseError{
				Command: commandName,
				Success: false,
				Message: response.Error,
			}
		}
		return response, nil
	}
}

func (d *PiRPCDriver) preflightResumedSession(ctx context.Context, sessionID string) error {
	response, err := d.requestCommand(ctx, map[string]any{"type": "get_state"})
	if err != nil {
		return err
	}
	if strings.TrimSpace(sessionID) == "" {
		return fmt.Errorf("pi get_state preflight requires a resumed session id")
	}
	if response.Success == nil || !*response.Success {
		return fmt.Errorf("pi get_state preflight returned an invalid response")
	}
	if len(response.Data) == 0 || strings.TrimSpace(string(response.Data)) == "null" {
		return fmt.Errorf("pi get_state preflight returned no session state")
	}
	var state struct {
		SessionID string `json:"sessionId"`
	}
	if err := json.Unmarshal(response.Data, &state); err != nil {
		return fmt.Errorf("decode pi get_state response: %w", err)
	}
	got := strings.TrimSpace(state.SessionID)
	want := strings.TrimSpace(sessionID)
	if got == "" {
		return fmt.Errorf("pi get_state response is missing session id")
	}
	if got != want {
		return fmt.Errorf("pi get_state returned session %q for requested session %q", got, want)
	}
	return nil
}

// sendCommand sends one RPC command and waits for its correlated response.
func (d *PiRPCDriver) sendCommand(ctx context.Context, command map[string]any) error {
	_, err := d.requestCommand(ctx, command)
	return err
}