package node_test

import (
	"context"
	"errors"
	"io"
	"net/http"
	"net/http/httptest"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/structpb"
	"iop/apps/node/internal/adapters"
	"iop/apps/node/internal/node"
	"iop/apps/node/internal/router"
	"iop/apps/node/internal/store"
	"iop/apps/node/internal/transport"
	runtime "iop/packages/go/execution"
	iop "iop/proto/gen/iop"
)

// --- admission gate / config refresh tests ---

// TestNodeConfigRefreshWithoutApplyManagerReportsRestartRequired verifies that
// a Node without a live apply manager responds restart_required for any refresh
// request that carries changed paths, and applied for a no-op (empty) request.
func TestNodeConfigRefreshWithoutApplyManagerReportsRestartRequired(t *testing.T) {
	router := &fixedRouter{adapterName: "test", adapters: map[string]runtime.Provider{"test": &countingAdapter{}}}
	n, _ := makeNode(t, router)

	// Non-empty changed_paths → restart_required.
	resp, err := n.OnConfigRefresh(context.Background(), nil, &iop.NodeConfigRefreshRequest{
		RequestId:    "req-restart",
		ChangedPaths: []string{"nodes.0.providers.0.capacity"},
	})
	if err != nil {
		t.Fatalf("OnConfigRefresh: %v", err)
	}
	if resp.GetStatus() != iop.NodeConfigRefreshStatus_NODE_CONFIG_REFRESH_STATUS_RESTART_REQUIRED {
		t.Fatalf("expected restart_required, got %v", resp.GetStatus())
	}
	if resp.GetRequestId() != "req-restart" {
		t.Fatalf("expected request_id=req-restart, got %q", resp.GetRequestId())
	}
	if len(resp.GetRestartRequiredPaths()) == 0 {
		t.Fatal("expected restart_required_paths to be populated")
	}

	// No config and no changed_paths → applied (no-op).
	resp2, err := n.OnConfigRefresh(context.Background(), nil, &iop.NodeConfigRefreshRequest{
		RequestId: "req-noop",
	})
	if err != nil {
		t.Fatalf("OnConfigRefresh noop: %v", err)
	}
	if resp2.GetStatus() != iop.NodeConfigRefreshStatus_NODE_CONFIG_REFRESH_STATUS_APPLIED {
		t.Fatalf("expected applied for no-op, got %v", resp2.GetStatus())
	}
}

func TestNodeConfigRefreshAppliesOpenAICompatCapacity(t *testing.T) {
	initialPayload := &iop.NodeConfigPayload{
		Adapters: []*iop.AdapterConfig{
			{
				Name:    "openai",
				Type:    "openai_compat",
				Enabled: true,
				Config: &iop.AdapterConfig_OpenaiCompat{
					OpenaiCompat: &iop.OpenAICompatAdapterConfig{
						Provider: "openai",
						Endpoint: "http://localhost:8080",
						Capacity: 2,
					},
				},
			},
		},
	}

	set, err := adapters.BuildConfigSet(initialPayload, zap.NewNop())
	if err != nil {
		t.Fatalf("BuildConfigSet: %v", err)
	}

	rtr := router.New(set.Registry, zap.NewNop())
	st, err := store.New(":memory:", zap.NewNop())
	if err != nil {
		t.Fatalf("store: %v", err)
	}
	t.Cleanup(func() { _ = st.Close() })

	n := node.New("test-node", rtr, st, 1, io.Discard, zap.NewNop(), set)

	// Check initial capabilities
	resp, err := n.OnCommandRequest(context.Background(), nil, &iop.NodeCommandRequest{
		RequestId: "cmd-1",
		Type:      iop.NodeCommandType_NODE_COMMAND_TYPE_CAPABILITIES,
		Adapter:   "openai",
	})
	if err != nil {
		t.Fatalf("OnCommandRequest: %v", err)
	}
	if resp.GetResult()["capacity"] != "2" {
		t.Fatalf("expected capacity 2, got %s", resp.GetResult()["capacity"])
	}

	// Refresh payload with new capacity
	refreshPayload := &iop.NodeConfigPayload{
		Adapters: []*iop.AdapterConfig{
			{
				Name:    "openai",
				Type:    "openai_compat",
				Enabled: true,
				Config: &iop.AdapterConfig_OpenaiCompat{
					OpenaiCompat: &iop.OpenAICompatAdapterConfig{
						Provider: "openai",
						Endpoint: "http://localhost:8080",
						Capacity: 5,
					},
				},
			},
		},
	}

	refreshResp, err := n.OnConfigRefresh(context.Background(), nil, &iop.NodeConfigRefreshRequest{
		RequestId: "refresh-1",
		Config:    refreshPayload,
	})
	if err != nil {
		t.Fatalf("OnConfigRefresh: %v", err)
	}
	if refreshResp.GetStatus() != iop.NodeConfigRefreshStatus_NODE_CONFIG_REFRESH_STATUS_APPLIED {
		t.Fatalf("expected applied status, got %v", refreshResp.GetStatus())
	}

	// Check updated capabilities
	resp2, err := n.OnCommandRequest(context.Background(), nil, &iop.NodeCommandRequest{
		RequestId: "cmd-2",
		Type:      iop.NodeCommandType_NODE_COMMAND_TYPE_CAPABILITIES,
		Adapter:   "openai",
	})
	if err != nil {
		t.Fatalf("OnCommandRequest: %v", err)
	}
	if resp2.GetResult()["capacity"] != "5" {
		t.Fatalf("expected capacity 5, got %s", resp2.GetResult()["capacity"])
	}
}

func TestNodeConfigRefreshUsesUpdatedOpenAICompatEndpoint(t *testing.T) {
	var oldCalls int32
	var newCalls int32

	serverOld := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		atomic.AddInt32(&oldCalls, 1)
		w.Header().Set("Content-Type", "application/json")
		_, _ = w.Write([]byte(`{"data": [{"id": "gpt-4o"}]}`))
	}))
	defer serverOld.Close()

	serverNew := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		atomic.AddInt32(&newCalls, 1)
		w.Header().Set("Content-Type", "application/json")
		_, _ = w.Write([]byte(`{"data": [{"id": "gpt-4o"}]}`))
	}))
	defer serverNew.Close()

	initialPayload := &iop.NodeConfigPayload{
		Adapters: []*iop.AdapterConfig{
			{
				Name:    "openai",
				Type:    "openai_compat",
				Enabled: true,
				Config: &iop.AdapterConfig_OpenaiCompat{
					OpenaiCompat: &iop.OpenAICompatAdapterConfig{
						Provider: "openai",
						Endpoint: serverOld.URL,
						Capacity: 2,
					},
				},
			},
		},
	}

	set, err := adapters.BuildConfigSet(initialPayload, zap.NewNop())
	if err != nil {
		t.Fatalf("BuildConfigSet: %v", err)
	}

	rtr := router.New(set.Registry, zap.NewNop())
	st, err := store.New(":memory:", zap.NewNop())
	if err != nil {
		t.Fatalf("store: %v", err)
	}
	t.Cleanup(func() { _ = st.Close() })

	n := node.New("test-node", rtr, st, 1, io.Discard, zap.NewNop(), set)

	// Call capabilities command -> triggers probe (requests serverOld)
	_, err = n.OnCommandRequest(context.Background(), nil, &iop.NodeCommandRequest{
		RequestId: "cmd-1",
		Type:      iop.NodeCommandType_NODE_COMMAND_TYPE_CAPABILITIES,
		Adapter:   "openai",
		Target:    "gpt-4o",
	})
	if err != nil {
		t.Fatalf("OnCommandRequest: %v", err)
	}

	if atomic.LoadInt32(&oldCalls) != 2 {
		t.Fatalf("expected 2 calls to serverOld, got %d", oldCalls)
	}
	if atomic.LoadInt32(&newCalls) != 0 {
		t.Fatalf("expected 0 calls to serverNew, got %d", newCalls)
	}

	// Refresh with new endpoint
	refreshPayload := &iop.NodeConfigPayload{
		Adapters: []*iop.AdapterConfig{
			{
				Name:    "openai",
				Type:    "openai_compat",
				Enabled: true,
				Config: &iop.AdapterConfig_OpenaiCompat{
					OpenaiCompat: &iop.OpenAICompatAdapterConfig{
						Provider: "openai",
						Endpoint: serverNew.URL,
						Capacity: 2,
					},
				},
			},
		},
	}

	refreshResp, err := n.OnConfigRefresh(context.Background(), nil, &iop.NodeConfigRefreshRequest{
		RequestId: "refresh-1",
		Config:    refreshPayload,
	})
	if err != nil {
		t.Fatalf("OnConfigRefresh: %v", err)
	}
	if refreshResp.GetStatus() != iop.NodeConfigRefreshStatus_NODE_CONFIG_REFRESH_STATUS_APPLIED {
		t.Fatalf("expected applied status, got %v", refreshResp.GetStatus())
	}

	// Call capabilities command again -> triggers probe (requests serverNew)
	_, err = n.OnCommandRequest(context.Background(), nil, &iop.NodeCommandRequest{
		RequestId: "cmd-2",
		Type:      iop.NodeCommandType_NODE_COMMAND_TYPE_CAPABILITIES,
		Adapter:   "openai",
		Target:    "gpt-4o",
	})
	if err != nil {
		t.Fatalf("OnCommandRequest 2: %v", err)
	}

	if atomic.LoadInt32(&oldCalls) != 2 {
		t.Fatalf("expected still 2 calls to serverOld, got %d", oldCalls)
	}
	if atomic.LoadInt32(&newCalls) != 2 {
		t.Fatalf("expected 2 calls to serverNew, got %d", newCalls)
	}
}

func TestNodeConfigRefreshUsesUpdatedOpenAICompatHeaders(t *testing.T) {
	var observedHeader string

	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		observedHeader = r.Header.Get("X-Test-Key")
		w.Header().Set("Content-Type", "application/json")
		_, _ = w.Write([]byte(`{"data": [{"id": "gpt-4o"}]}`))
	}))
	defer server.Close()

	initialPayload := &iop.NodeConfigPayload{
		Adapters: []*iop.AdapterConfig{
			{
				Name:    "openai",
				Type:    "openai_compat",
				Enabled: true,
				Config: &iop.AdapterConfig_OpenaiCompat{
					OpenaiCompat: &iop.OpenAICompatAdapterConfig{
						Provider: "openai",
						Endpoint: server.URL,
						Headers:  map[string]string{"X-Test-Key": "initial-value"},
						Capacity: 2,
					},
				},
			},
		},
	}

	set, err := adapters.BuildConfigSet(initialPayload, zap.NewNop())
	if err != nil {
		t.Fatalf("BuildConfigSet: %v", err)
	}

	rtr := router.New(set.Registry, zap.NewNop())
	st, err := store.New(":memory:", zap.NewNop())
	if err != nil {
		t.Fatalf("store: %v", err)
	}
	t.Cleanup(func() { _ = st.Close() })

	n := node.New("test-node", rtr, st, 1, io.Discard, zap.NewNop(), set)

	// Trigger request -> should see initial header
	_, _ = n.OnCommandRequest(context.Background(), nil, &iop.NodeCommandRequest{
		RequestId: "cmd-1",
		Type:      iop.NodeCommandType_NODE_COMMAND_TYPE_CAPABILITIES,
		Adapter:   "openai",
		Target:    "gpt-4o",
	})
	if observedHeader != "initial-value" {
		t.Fatalf("expected header 'initial-value', got %q", observedHeader)
	}

	// Refresh with new header
	refreshPayload := &iop.NodeConfigPayload{
		Adapters: []*iop.AdapterConfig{
			{
				Name:    "openai",
				Type:    "openai_compat",
				Enabled: true,
				Config: &iop.AdapterConfig_OpenaiCompat{
					OpenaiCompat: &iop.OpenAICompatAdapterConfig{
						Provider: "openai",
						Endpoint: server.URL,
						Headers:  map[string]string{"X-Test-Key": "updated-value"},
						Capacity: 2,
					},
				},
			},
		},
	}

	_, _ = n.OnConfigRefresh(context.Background(), nil, &iop.NodeConfigRefreshRequest{
		RequestId: "refresh-1",
		Config:    refreshPayload,
	})

	// Trigger request again -> should see updated header
	_, _ = n.OnCommandRequest(context.Background(), nil, &iop.NodeCommandRequest{
		RequestId: "cmd-2",
		Type:      iop.NodeCommandType_NODE_COMMAND_TYPE_CAPABILITIES,
		Adapter:   "openai",
		Target:    "gpt-4o",
	})
	if observedHeader != "updated-value" {
		t.Fatalf("expected header 'updated-value', got %q", observedHeader)
	}
}

func TestNodeConfigRefreshUpdatesExistingAdapterGateCapacity(t *testing.T) {
	// mock HTTP server that can be blocked or unblocked
	blockChan := make(chan struct{})
	defer func() {
		select {
		case <-blockChan:
		default:
			close(blockChan)
		}
	}()

	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "application/json")
		if strings.Contains(r.URL.Path, "/v1/models") {
			_, _ = w.Write([]byte(`{"data": [{"id": "gpt-4o"}]}`))
			return
		}
		// Block completion requests until blockChan is closed or read
		<-blockChan
		// Write a dummy SSE chat completion response
		_, _ = w.Write([]byte("data: {\"choices\": [{\"delta\": {\"content\": \"hello\"}}]}\n\ndata: [DONE]\n\n"))
	}))
	defer server.Close()

	initialPayload := &iop.NodeConfigPayload{
		Adapters: []*iop.AdapterConfig{
			{
				Name:    "openai",
				Type:    "openai_compat",
				Enabled: true,
				Config: &iop.AdapterConfig_OpenaiCompat{
					OpenaiCompat: &iop.OpenAICompatAdapterConfig{
						Provider: "openai",
						Endpoint: server.URL,
						Capacity: 1,
					},
				},
			},
		},
	}

	set, err := adapters.BuildConfigSet(initialPayload, zap.NewNop())
	if err != nil {
		t.Fatalf("BuildConfigSet: %v", err)
	}

	rtr := router.New(set.Registry, zap.NewNop())
	st, err := store.New(":memory:", zap.NewNop())
	if err != nil {
		t.Fatalf("store: %v", err)
	}
	t.Cleanup(func() { _ = st.Close() })

	n := node.New("test-node", rtr, st, 0, io.Discard, zap.NewNop(), set)

	inputMap := map[string]interface{}{
		"prompt": "hello",
	}
	inputStruct, err := structpb.NewStruct(inputMap)
	if err != nil {
		t.Fatalf("failed to create input struct: %v", err)
	}

	// First request: should run and block inside the mock server
	errChan1 := make(chan error, 1)
	go func() {
		errChan1 <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
			RunId:      "run-1",
			Adapter:    "openai",
			Target:     "gpt-4o",
			Background: false,
			Input:      inputStruct,
		})
	}()

	// Wait a bit to ensure the first request is indeed running and blocked
	time.Sleep(50 * time.Millisecond)

	// Second request: should fail because capacity is 1
	err2 := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
		RunId:      "run-2",
		Adapter:    "openai",
		Target:     "gpt-4o",
		Background: false,
		Input:      inputStruct,
	})
	if err2 == nil {
		t.Fatal("expected second request to fail due to concurrency limit")
	}
	if !errors.Is(err2, node.ErrConcurrencyLimitExceeded) {
		t.Fatalf("expected ErrConcurrencyLimitExceeded, got: %v", err2)
	}

	// Refresh config to increase capacity to 2
	refreshPayload := &iop.NodeConfigPayload{
		Adapters: []*iop.AdapterConfig{
			{
				Name:    "openai",
				Type:    "openai_compat",
				Enabled: true,
				Config: &iop.AdapterConfig_OpenaiCompat{
					OpenaiCompat: &iop.OpenAICompatAdapterConfig{
						Provider: "openai",
						Endpoint: server.URL,
						Capacity: 2,
					},
				},
			},
		},
	}

	refreshResp, refreshErr := n.OnConfigRefresh(context.Background(), nil, &iop.NodeConfigRefreshRequest{
		RequestId: "refresh-1",
		Config:    refreshPayload,
	})
	if refreshErr != nil {
		t.Fatalf("OnConfigRefresh failed: %v", refreshErr)
	}
	if refreshResp.Status != iop.NodeConfigRefreshStatus_NODE_CONFIG_REFRESH_STATUS_APPLIED {
		t.Fatalf("expected refresh status APPLIED, got %v", refreshResp.Status)
	}

	// Third request: should now succeed (and block) instead of getting rejected
	errChan3 := make(chan error, 1)
	go func() {
		errChan3 <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
			RunId:      "run-3",
			Adapter:    "openai",
			Target:     "gpt-4o",
			Background: false,
			Input:      inputStruct,
		})
	}()

	// Wait a bit and check that third request hasn't errored immediately (it should be blocking)
	time.Sleep(50 * time.Millisecond)
	select {
	case err3 := <-errChan3:
		t.Fatalf("expected third request to block, but it failed immediately: %v", err3)
	default:
		// It is blocking as expected!
	}

	// Unblock mock server for all requests
	close(blockChan)

	// Wait for run-1 and run-3 to finish successfully
	if err1 := <-errChan1; err1 != nil {
		t.Fatalf("run-1 failed: %v", err1)
	}
	if err3 := <-errChan3; err3 != nil {
		t.Fatalf("run-3 failed: %v", err3)
	}
}

func TestNodeConfigRefreshDecreasesExistingAdapterGateCapacity(t *testing.T) {
	blockChan := make(chan struct{})
	defer func() {
		select {
		case <-blockChan:
		default:
			close(blockChan)
		}
	}()

	startedChan := make(chan struct{}, 1)

	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "application/json")
		if strings.Contains(r.URL.Path, "/v1/models") {
			_, _ = w.Write([]byte(`{"data": [{"id": "gpt-4o"}]}`))
			return
		}
		select {
		case startedChan <- struct{}{}:
		default:
		}
		<-blockChan
		_, _ = w.Write([]byte("data: {\"choices\": [{\"delta\": {\"content\": \"hello\"}}]}\n\ndata: [DONE]\n\n"))
	}))
	defer server.Close()

	initialPayload := &iop.NodeConfigPayload{
		Adapters: []*iop.AdapterConfig{
			{
				Name:    "openai",
				Type:    "openai_compat",
				Enabled: true,
				Config: &iop.AdapterConfig_OpenaiCompat{
					OpenaiCompat: &iop.OpenAICompatAdapterConfig{
						Provider: "openai",
						Endpoint: server.URL,
						Capacity: 2,
					},
				},
			},
		},
	}

	set, err := adapters.BuildConfigSet(initialPayload, zap.NewNop())
	if err != nil {
		t.Fatalf("BuildConfigSet: %v", err)
	}

	rtr := router.New(set.Registry, zap.NewNop())
	st, err := store.New(":memory:", zap.NewNop())
	if err != nil {
		t.Fatalf("store: %v", err)
	}
	t.Cleanup(func() { _ = st.Close() })

	n := node.New("test-node", rtr, st, 0, io.Discard, zap.NewNop(), set)

	inputMap := map[string]interface{}{
		"prompt": "hello",
	}
	inputStruct, err := structpb.NewStruct(inputMap)
	if err != nil {
		t.Fatalf("failed to create input struct: %v", err)
	}

	errChan1 := make(chan error, 1)
	go func() {
		errChan1 <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
			RunId:      "run-1",
			Adapter:    "openai",
			Target:     "gpt-4o",
			Background: false,
			Input:      inputStruct,
		})
	}()

	<-startedChan

	refreshPayload := &iop.NodeConfigPayload{
		Adapters: []*iop.AdapterConfig{
			{
				Name:    "openai",
				Type:    "openai_compat",
				Enabled: true,
				Config: &iop.AdapterConfig_OpenaiCompat{
					OpenaiCompat: &iop.OpenAICompatAdapterConfig{
						Provider: "openai",
						Endpoint: server.URL,
						Capacity: 1,
					},
				},
			},
		},
	}

	refreshResp, refreshErr := n.OnConfigRefresh(context.Background(), nil, &iop.NodeConfigRefreshRequest{
		RequestId: "refresh-1",
		Config:    refreshPayload,
	})
	if refreshErr != nil {
		t.Fatalf("OnConfigRefresh failed: %v", refreshErr)
	}
	if refreshResp.Status != iop.NodeConfigRefreshStatus_NODE_CONFIG_REFRESH_STATUS_APPLIED {
		t.Fatalf("expected refresh status APPLIED, got %v", refreshResp.Status)
	}

	ctx2, cancel2 := context.WithTimeout(context.Background(), 200*time.Millisecond)
	defer cancel2()
	err2 := n.OnRunRequest(ctx2, &transport.Session{}, &iop.RunRequest{
		RunId:      "run-2",
		Adapter:    "openai",
		Target:     "gpt-4o",
		Background: false,
		Input:      inputStruct,
	})
	if err2 == nil {
		t.Fatal("expected second request to fail due to concurrency limit")
	}
	if !errors.Is(err2, node.ErrConcurrencyLimitExceeded) {
		t.Fatalf("expected ErrConcurrencyLimitExceeded, got: %v", err2)
	}

	close(blockChan)

	if err1 := <-errChan1; err1 != nil {
		t.Fatalf("run-1 failed: %v", err1)
	}
}

// --- runtime concurrency metadata tests (admission is no-op) ---

func TestConfigRefreshRuntimeConcurrencyDoesNotAffectAdmission(t *testing.T) {
	// Adapter is unlimited (MaxConcurrency=0), so the adapter gate imposes no limit.
	sa := newQueuedSlowAdapter("slow", 0, 0, 0)
	rtr := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Provider{"slow": sa}}
	st, err := store.New(":memory:", zap.NewNop())
	if err != nil {
		t.Fatalf("store: %v", err)
	}
	t.Cleanup(func() { _ = st.Close() })

	// Global concurrency is set to 1, but it must NOT affect admission.
	// All concurrent runs should succeed because the adapter gate is unlimited.
	n := node.New("test-node", rtr, st, 1, io.Discard, zap.NewNop(), nil)

	errCh1 := make(chan error, 1)
	go func() {
		errCh1 <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{RunId: "run-1", Adapter: "slow"})
	}()
	waitStarted(t, sa, "run-1")

	// Second run should NOT be rejected: global concurrency is no longer used for admission.
	errCh2 := make(chan error, 1)
	go func() {
		errCh2 <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{RunId: "run-2", Adapter: "slow"})
	}()
	waitStarted(t, sa, "run-2")

	// Config refresh with runtime concurrency must succeed and store metadata.
	resp, refreshErr := n.OnConfigRefresh(context.Background(), nil, &iop.NodeConfigRefreshRequest{
		RequestId: "refresh-conc",
		Config:    &iop.NodeConfigPayload{Runtime: &iop.NodeRuntimeConfig{Concurrency: 2}},
	})
	if refreshErr != nil {
		t.Fatalf("OnConfigRefresh failed: %v", refreshErr)
	}
	if resp.Status != iop.NodeConfigRefreshStatus_NODE_CONFIG_REFRESH_STATUS_APPLIED {
		t.Fatalf("expected refresh status APPLIED, got %v", resp.Status)
	}

	// Third run must also succeed: runtime concurrency is metadata only.
	errCh3 := make(chan error, 1)
	go func() {
		errCh3 <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{RunId: "run-3", Adapter: "slow"})
	}()
	waitStarted(t, sa, "run-3")

	sa.releaseRun("run-1")
	sa.releaseRun("run-2")
	sa.releaseRun("run-3")
	if err := <-errCh1; err != nil {
		t.Fatalf("run-1 failed: %v", err)
	}
	if err := <-errCh2; err != nil {
		t.Fatalf("run-2 failed: %v", err)
	}
	if err := <-errCh3; err != nil {
		t.Fatalf("run-3 failed: %v", err)
	}
}

func TestConfigRefreshConcurrencyDecreaseDoesNotAffectAdmission(t *testing.T) {
	sa := newQueuedSlowAdapter("slow", 0, 0, 0)
	rtr := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Provider{"slow": sa}}
	st, err := store.New(":memory:", zap.NewNop())
	if err != nil {
		t.Fatalf("store: %v", err)
	}
	t.Cleanup(func() { _ = st.Close() })

	n := node.New("test-node", rtr, st, 2, io.Discard, zap.NewNop(), nil)

	errCh1 := make(chan error, 1)
	go func() {
		errCh1 <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{RunId: "run-1", Adapter: "slow"})
	}()
	waitStarted(t, sa, "run-1")

	errCh2 := make(chan error, 1)
	go func() {
		errCh2 <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{RunId: "run-2", Adapter: "slow"})
	}()
	waitStarted(t, sa, "run-2")

	// Config refresh decreases runtime concurrency from 2 to 1.
	// This must NOT affect admission: the adapter gate is unlimited.
	resp, refreshErr := n.OnConfigRefresh(context.Background(), nil, &iop.NodeConfigRefreshRequest{
		RequestId: "refresh-dec",
		Config:    &iop.NodeConfigPayload{Runtime: &iop.NodeRuntimeConfig{Concurrency: 1}},
	})
	if refreshErr != nil {
		t.Fatalf("OnConfigRefresh failed: %v", refreshErr)
	}
	if resp.Status != iop.NodeConfigRefreshStatus_NODE_CONFIG_REFRESH_STATUS_APPLIED {
		t.Fatalf("expected refresh status APPLIED, got %v", resp.Status)
	}

	// Third run must be admitted: global concurrency is no longer used for admission.
	errCh3 := make(chan error, 1)
	go func() {
		errCh3 <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{RunId: "run-3", Adapter: "slow"})
	}()
	waitStarted(t, sa, "run-3")

	// All three runs complete normally.
	sa.releaseRun("run-1")
	sa.releaseRun("run-2")
	sa.releaseRun("run-3")
	if err := <-errCh1; err != nil {
		t.Fatalf("run-1 should complete normally, got %v", err)
	}
	if err := <-errCh2; err != nil {
		t.Fatalf("run-2 should complete normally, got %v", err)
	}
	if err := <-errCh3; err != nil {
		t.Fatalf("run-3 should complete normally, got %v", err)
	}
}
