package dispatcher

import (
	"context"
	"os"
	"path/filepath"
	"runtime"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/desktop/internal/agent"
	"github.com/tidbcloud/AgentTeam/desktop/internal/agent/drivercontract"
	"github.com/tidbcloud/AgentTeam/desktop/internal/agent/runtimeconfig"
	"github.com/tidbcloud/AgentTeam/desktop/internal/types"
)

// writePiReconcileIntegrationBinary is intentionally a tiny RPC fixture rather
// than a reconciliation result stub. The Process created by Dispatcher uses a
// real PiRPCDriver; this fixture only supplies the child transport and emits a
// q2 user frame after q1's restored owner has failed closed.
func writePiReconcileIntegrationBinary(t *testing.T) string {
	t.Helper()
	dir := t.TempDir()
	calls := filepath.Join(dir, "calls.jsonl")
	script := filepath.Join(dir, "pi")
	if runtime.GOOS == "windows" {
		script += ".cmd"
		require.NoError(t, os.WriteFile(script, []byte(
			"@echo off\r\n"+
				"for /f \"delims=\" %%L in ('more') do (\r\n"+
				"  >>\"%PI_TEST_PI_CALLS%\" echo %%L\r\n"+
				"  echo %%L | findstr /c:\"\\\"type\\\":\\\"prompt\\\"\" >nul\r\n"+
				"  if not errorlevel 1 (\r\n"+
				"    echo {\"type\":\"response\",\"id\":1,\"command\":\"prompt\",\"success\":true}\r\n"+
				"    echo {\"type\":\"message_start\",\"message\":{\"role\":\"user\",\"content\":\"q2\\n\\n^<!-- agentteam-runtime-input:q2 -->\"}}\r\n"+
				"    echo {\"type\":\"agent_settled\"}\r\n"+
				"  )\r\n"+
				")\r\n",
		), 0o755))
	} else {
		require.NoError(t, os.WriteFile(script, []byte(
			"#!/bin/sh\n"+
				"while IFS= read -r line; do\n"+
				"  printf '%s\\n' \"$line\" >> \"$PI_TEST_PI_CALLS\"\n"+
				"  case \"$line\" in\n"+
				"    *'\"type\":\"prompt\"'*)\n"+
				"      id=$(printf '%s' \"$line\" | sed -n 's/.*\"id\":\\([0-9][0-9]*\\).*/\\1/p')\n"+
				"      [ -n \"$id\" ] || id=1\n"+
				"      printf '%s\\n' \"{\\\"type\\\":\\\"response\\\",\\\"id\\\":$id,\\\"command\\\":\\\"prompt\\\",\\\"success\\\":true}\"\n"+
				"      printf '%s\\n' '{\"type\":\"message_start\",\"message\":{\"role\":\"user\",\"content\":\"q2\\n\\n<!-- agentteam-runtime-input:q2 -->\"}}'\n"+
				"      printf '%s\\n' '{\"type\":\"agent_settled\"}'\n"+
				"      ;;\n"+
				"  esac\n"+
				"done\n",
		), 0o755))
	}
	t.Setenv("PI_TEST_PI_CALLS", calls)
	return script
}

func TestDispatcherPiRestartBeforeMarkerUsesRealReconcileAndFailsClosed(t *testing.T) {
	t.Setenv("AGENTTEAM_CONFIG_DIR", t.TempDir())
	binary := writePiReconcileIntegrationBinary(t)
	d := mustNewDispatcher("http://server.test")
	// Register dispatcher shutdown after the Pi fixture's TempDir cleanup so
	// LIFO cleanup stops and joins the child process before removing its files.
	t.Cleanup(d.Close)

	const (
		agentID   = "pi-reconcile-dispatcher"
		sessionID = "00000000-0000-0000-0000-000000000001"
	)
	sessionKey := sessionKeyForBinding(agentID, conversationBinding{}).String()
	workDir := sessionWorkDirForKey(agentID, sessionKey, false)
	runtimeStateDir := agentRuntimeStateDir(agentID)
	sessionDir := filepath.Join(runtimeStateDir, ".pi-sessions")
	require.NoError(t, os.MkdirAll(sessionDir, 0o700))
	sessionPath := filepath.Join(sessionDir, "2026_"+sessionID+".jsonl")
	// Existing readable JSONL without q1's marker is appendable history, not a
	// negative fence. ReconcileRuntimeInput must therefore return unknown.
	require.NoError(t, os.WriteFile(sessionPath, []byte(`{"type":"session","id":"`+sessionID+`"}`+"\n"), 0o600))

	acceptedAt := time.Now().UTC().Add(-time.Minute)
	require.NoError(t, saveSessionResumeState(workDir, sessionResumeState{
		LastSessionID:                   sessionID,
		LastStartConfig:                 &types.AgentStartConfig{SessionID: sessionID, Runtime: "pi", PermissionMode: types.AgentPermissionModeFullAccess},
		RuntimeDeliveryOwnershipVersion: runtimeDeliveryOwnershipVersion,
		RuntimeDeliveryOwnership: []agent.DeliveryOwnershipRecord{
			{
				Message:        types.InboxMessage{MessageID: "q1", TargetKey: "thread-1", TargetSeq: 1, Target: "#ops", Content: "q1 must never replay"},
				Input:          drivercontract.RuntimeInput{ID: "q1"},
				SessionID:      sessionID,
				State:          agent.DeliveryOwnershipInflight,
				InflightReason: agent.DeliveryInflightDispatching,
				AcceptedAt:     acceptedAt,
			},
			{
				Message:    types.InboxMessage{MessageID: "q2", TargetKey: "thread-1", TargetSeq: 2, Target: "#ops", Content: "q2 advances after q1 terminal"},
				Input:      drivercontract.RuntimeInput{ID: "q2"},
				State:      agent.DeliveryOwnershipQueued,
				AcceptedAt: acceptedAt.Add(time.Second),
			},
		},
	}))

	_, err := d.deliverResolvedSession(
		context.Background(),
		sessionKey,
		agent.Config{ID: agentID, Runtime: "pi", PermissionMode: types.AgentPermissionModeFullAccess},
		runtimeconfig.Runtime{
			Name:   "pi",
			Type:   runtimeconfig.TypePi,
			Binary: binary,
		},
		conversationBinding{channelID: "ops", serverAgentID: agentID},
		types.InboxMessage{},
		startSessionOptions{},
	)
	require.NoError(t, err)

	require.Eventually(t, func() bool {
		state, loadErr := loadSessionResumeState(workDir)
		if loadErr != nil || len(state.RuntimeDeliveryOwnership) != 2 {
			return false
		}
		var q1, q2 agent.DeliveryOwnershipRecord
		for _, record := range state.RuntimeDeliveryOwnership {
			switch record.MessageID() {
			case "q1":
				q1 = record
			case "q2":
				q2 = record
			}
		}
		return q1.IsFailed() &&
			q1.FailureCode == "runtime_input_reconciliation_exhausted" &&
			q2.IsAcked()
	}, 12*time.Second, 20*time.Millisecond)

	calls, err := os.ReadFile(os.Getenv("PI_TEST_PI_CALLS"))
	require.NoError(t, err)
	callLog := string(calls)
	require.NotContains(t, callLog, "agentteam-runtime-input:q1", "restored q1 body must never be replayed")
	require.Contains(t, callLog, "agentteam-runtime-input:q2", "q2 advances only after q1's durable failed terminal")
}