package automation

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"testing"

	"github.com/nexuscrm/shared/pkg/models"
)

// fakeDirectoryReader is a test double for ConnectorDirectoryReader.
type fakeDirectoryReader struct {
	items       []map[string]interface{}
	err         error
	capturedReq ConnectorDirectoryPullRequest
}

func (f *fakeDirectoryReader) PullDirectory(_ context.Context, req ConnectorDirectoryPullRequest) ([]map[string]interface{}, error) {
	f.capturedReq = req
	return f.items, f.err
}

// makeFlowVersionWithSource builds a minimal FlowVersion with a ConnectorSource.
func makeFlowVersionWithSource(src *models.FlowConnectorSource) *models.FlowVersion {
	return &models.FlowVersion{
		DefinitionOwnerPack: "test-pack",
		Source:              flowSourceStandard, // pack-owned standard (uncustomized)
		Metadata: models.FlowMetadata{
			Start: &models.FlowStart{
				ConnectorSource: src,
			},
		},
	}
}

// TestBuildConnectorDirectoryInputs_CustomizedOrNonStandard_ReturnsError verifies
// that a tenant-customized or non-standard-source flow version (which keeps its
// OwnerPack) cannot use pack-owned connector profile authority — matching the
// flowVersionIsPackOwnedStandard contract connector push enforces.
func TestBuildConnectorDirectoryInputs_CustomizedOrNonStandard_ReturnsError(t *testing.T) {
	exec := &FlowExecutor{
		connectorDirectoryReader: &fakeDirectoryReader{items: []map[string]interface{}{{"id": "x1"}}},
	}

	customized := makeFlowVersionWithSource(validSrc())
	customized.DefinitionIsCustomized = true
	if _, err := exec.buildConnectorDirectoryInputs(context.Background(), customized, nil); err == nil {
		t.Fatal("expected error for customized flow version, got nil")
	}

	nonStandard := makeFlowVersionWithSource(validSrc())
	nonStandard.Source = "custom"
	if _, err := exec.buildConnectorDirectoryInputs(context.Background(), nonStandard, nil); err == nil {
		t.Fatal("expected error for non-standard source flow version, got nil")
	}
}

func validSrc() *models.FlowConnectorSource {
	return &models.FlowConnectorSource{
		OwnerPack:   "test-pack",
		ProfileName: "directory-profile",
		ObjectKey:   "contacts",
		ItemsPath:   "items",
		ItemIDPath:  "id",
	}
}

// ---------------------------------------------------------------------------
// buildConnectorDirectoryInputs unit tests
// ---------------------------------------------------------------------------

func TestBuildConnectorDirectoryInputs_NItems_ReturnsNInputSets(t *testing.T) {
	items := []map[string]interface{}{
		{"id": "abc-1", "name": "Alice"},
		{"id": "abc-2", "name": "Bob"},
		{"id": "abc-3", "name": "Carol"},
	}
	fake := &fakeDirectoryReader{items: items}
	exec := &FlowExecutor{
		connectorDirectoryReader: fake,
	}
	fv := makeFlowVersionWithSource(validSrc())
	sets, err := exec.buildConnectorDirectoryInputs(context.Background(), fv, nil)
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if len(sets) != 3 {
		t.Fatalf("expected 3 input sets, got %d", len(sets))
	}
	// Each set must have a non-empty top-level "Id" and a "$Record".
	seen := make(map[string]bool)
	for i, s := range sets {
		id, ok := s["Id"].(string)
		if !ok || id == "" {
			t.Errorf("set %d: missing or empty top-level Id", i)
		}
		if _, ok := s["$Record"]; !ok {
			t.Errorf("set %d: missing $Record", i)
		}
		// scheduledInputRecordID must return the same non-empty, unique value.
		key := scheduledInputRecordID(s)
		if key == "" {
			t.Errorf("set %d: scheduledInputRecordID returned empty string", i)
		}
		if seen[key] {
			t.Errorf("set %d: duplicate scheduledInputRecordID %q", i, key)
		}
		seen[key] = true
	}
	// Verify the builder threads RequesterOwnerPack from DefinitionOwnerPack.
	if fake.capturedReq.RequesterOwnerPack != fv.DefinitionOwnerPack {
		t.Errorf("RequesterOwnerPack = %q; want %q", fake.capturedReq.RequesterOwnerPack, fv.DefinitionOwnerPack)
	}
	// Verify MaxItems is threaded through.
	if fake.capturedReq.MaxItems != scheduledFlowMaxMatched {
		t.Errorf("MaxItems = %d; want %d", fake.capturedReq.MaxItems, scheduledFlowMaxMatched)
	}
}

func TestBuildConnectorDirectoryInputs_MissingItemID_ReturnsError(t *testing.T) {
	// Item has no "id" field at ItemIDPath.
	items := []map[string]interface{}{
		{"name": "Alice"}, // no "id"
	}
	exec := &FlowExecutor{
		connectorDirectoryReader: &fakeDirectoryReader{items: items},
	}
	fv := makeFlowVersionWithSource(validSrc())
	_, err := exec.buildConnectorDirectoryInputs(context.Background(), fv, nil)
	if err == nil {
		t.Fatal("expected error for item with missing id, got nil")
	}
}

func TestBuildConnectorDirectoryInputs_EmptyItemID_ReturnsError(t *testing.T) {
	// Item has "id" = "" (empty string).
	items := []map[string]interface{}{
		{"id": "", "name": "Alice"},
	}
	exec := &FlowExecutor{
		connectorDirectoryReader: &fakeDirectoryReader{items: items},
	}
	fv := makeFlowVersionWithSource(validSrc())
	_, err := exec.buildConnectorDirectoryInputs(context.Background(), fv, nil)
	if err == nil {
		t.Fatal("expected error for item with empty id, got nil")
	}
}

func TestBuildConnectorDirectoryInputs_ReaderError_PropagatesError(t *testing.T) {
	pullErr := errors.New("network timeout")
	exec := &FlowExecutor{
		connectorDirectoryReader: &fakeDirectoryReader{err: pullErr},
	}
	fv := makeFlowVersionWithSource(validSrc())
	_, err := exec.buildConnectorDirectoryInputs(context.Background(), fv, nil)
	if err == nil {
		t.Fatal("expected error from reader, got nil")
	}
	if !errors.Is(err, pullErr) {
		t.Errorf("expected errors.Is(err, pullErr); got: %v", err)
	}
}

func TestBuildConnectorDirectoryInputs_EmptyDirectory_ReturnsEmptySlice(t *testing.T) {
	exec := &FlowExecutor{
		connectorDirectoryReader: &fakeDirectoryReader{items: []map[string]interface{}{}},
	}
	fv := makeFlowVersionWithSource(validSrc())
	sets, err := exec.buildConnectorDirectoryInputs(context.Background(), fv, nil)
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if len(sets) != 0 {
		t.Fatalf("expected 0 input sets, got %d", len(sets))
	}
}

func TestBuildConnectorDirectoryInputs_ExceedsCap_ReturnsError(t *testing.T) {
	// Build scheduledFlowMaxMatched+1 items.
	items := make([]map[string]interface{}, scheduledFlowMaxMatched+1)
	for i := range items {
		items[i] = map[string]interface{}{"id": fmt.Sprintf("id-%d", i)}
	}
	exec := &FlowExecutor{
		connectorDirectoryReader: &fakeDirectoryReader{items: items},
	}
	fv := makeFlowVersionWithSource(validSrc())
	_, err := exec.buildConnectorDirectoryInputs(context.Background(), fv, nil)
	if err == nil {
		t.Fatal("expected error for item count exceeding cap, got nil")
	}
}

func TestBuildConnectorDirectoryInputs_MisconfiguredSource_BlankOwnerPack_ReturnsError(t *testing.T) {
	src := validSrc()
	src.OwnerPack = ""
	exec := &FlowExecutor{
		connectorDirectoryReader: &fakeDirectoryReader{},
	}
	fv := makeFlowVersionWithSource(src)
	_, err := exec.buildConnectorDirectoryInputs(context.Background(), fv, nil)
	if err == nil {
		t.Fatal("expected error for blank OwnerPack, got nil")
	}
}

func TestBuildConnectorDirectoryInputs_MisconfiguredSource_BlankProfileName_ReturnsError(t *testing.T) {
	src := validSrc()
	src.ProfileName = ""
	exec := &FlowExecutor{connectorDirectoryReader: &fakeDirectoryReader{}}
	fv := makeFlowVersionWithSource(src)
	_, err := exec.buildConnectorDirectoryInputs(context.Background(), fv, nil)
	if err == nil {
		t.Fatal("expected error for blank ProfileName, got nil")
	}
}

func TestBuildConnectorDirectoryInputs_MisconfiguredSource_BlankObjectKey_ReturnsError(t *testing.T) {
	src := validSrc()
	src.ObjectKey = ""
	exec := &FlowExecutor{connectorDirectoryReader: &fakeDirectoryReader{}}
	fv := makeFlowVersionWithSource(src)
	_, err := exec.buildConnectorDirectoryInputs(context.Background(), fv, nil)
	if err == nil {
		t.Fatal("expected error for blank ObjectKey, got nil")
	}
}

func TestBuildConnectorDirectoryInputs_MisconfiguredSource_BlankItemsPath_ReturnsError(t *testing.T) {
	src := validSrc()
	src.ItemsPath = ""
	exec := &FlowExecutor{connectorDirectoryReader: &fakeDirectoryReader{}}
	fv := makeFlowVersionWithSource(src)
	_, err := exec.buildConnectorDirectoryInputs(context.Background(), fv, nil)
	if err == nil {
		t.Fatal("expected error for blank ItemsPath, got nil")
	}
}

func TestBuildConnectorDirectoryInputs_MisconfiguredSource_BlankItemIDPath_ReturnsError(t *testing.T) {
	src := validSrc()
	src.ItemIDPath = ""
	exec := &FlowExecutor{connectorDirectoryReader: &fakeDirectoryReader{}}
	fv := makeFlowVersionWithSource(src)
	_, err := exec.buildConnectorDirectoryInputs(context.Background(), fv, nil)
	if err == nil {
		t.Fatal("expected error for blank ItemIDPath, got nil")
	}
}

func TestBuildConnectorDirectoryInputs_NilReader_ReturnsError(t *testing.T) {
	exec := &FlowExecutor{
		connectorDirectoryReader: nil,
	}
	fv := makeFlowVersionWithSource(validSrc())
	_, err := exec.buildConnectorDirectoryInputs(context.Background(), fv, nil)
	if err == nil {
		t.Fatal("expected error when connectorDirectoryReader is nil, got nil")
	}
}

// TestBuildConnectorDirectoryInputs_EmptyDefinitionOwnerPack_ReturnsError checks
// that a flow without a pack identity is rejected before the reader is called.
func TestBuildConnectorDirectoryInputs_EmptyDefinitionOwnerPack_ReturnsError(t *testing.T) {
	exec := &FlowExecutor{
		connectorDirectoryReader: &fakeDirectoryReader{},
	}
	fv := makeFlowVersionWithSource(validSrc())
	fv.DefinitionOwnerPack = "" // unset — simulates non-pack-owned flow
	_, err := exec.buildConnectorDirectoryInputs(context.Background(), fv, nil)
	if err == nil {
		t.Fatal("expected error for empty DefinitionOwnerPack, got nil")
	}
}

// ---------------------------------------------------------------------------
// buildScheduledRunInputs dispatch tests
// ---------------------------------------------------------------------------

func TestBuildScheduledRunInputs_ConnectorSource_DispatchesToDirectoryBuilder(t *testing.T) {
	items := []map[string]interface{}{
		{"id": "x1", "email": "a@example.com"},
	}
	exec := &FlowExecutor{
		connectorDirectoryReader: &fakeDirectoryReader{items: items},
	}
	fv := makeFlowVersionWithSource(validSrc())
	sets, err := exec.buildScheduledRunInputs(context.Background(), fv, nil)
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if len(sets) != 1 {
		t.Fatalf("expected 1 input set, got %d", len(sets))
	}
	if id, _ := sets[0]["Id"].(string); id != "x1" {
		t.Errorf("expected Id=x1, got %q", id)
	}
}

func TestBuildScheduledRunInputs_NoConnectorSource_NilStart_FallsThrough(t *testing.T) {
	// No ConnectorSource, no Object => single-nil-input path (global run).
	exec := &FlowExecutor{}
	fv := &models.FlowVersion{
		Metadata: models.FlowMetadata{
			Start: &models.FlowStart{},
		},
	}
	sets, err := exec.buildScheduledRunInputs(context.Background(), fv, nil)
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	// Expect the single-nil-input global-run slice.
	if len(sets) != 1 || sets[0] != nil {
		t.Fatalf("expected [{nil}], got %v", sets)
	}
}

// ---------------------------------------------------------------------------
// extractItemID helpers
// ---------------------------------------------------------------------------

func TestExtractItemID_TopLevelKey(t *testing.T) {
	item := map[string]interface{}{"id": "abc"}
	got, err := extractItemID(item, "id")
	if err != nil || got != "abc" {
		t.Errorf("expected abc, got %q, err=%v", got, err)
	}
}

func TestExtractItemID_NestedKey(t *testing.T) {
	item := map[string]interface{}{"meta": map[string]interface{}{"uid": "nested-id"}}
	got, err := extractItemID(item, "meta.uid")
	if err != nil || got != "nested-id" {
		t.Errorf("expected nested-id, got %q, err=%v", got, err)
	}
}

func TestExtractItemID_MissingKey_ReturnsEmpty(t *testing.T) {
	item := map[string]interface{}{"name": "Alice"}
	got, err := extractItemID(item, "id")
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if got != "" {
		t.Errorf("expected empty string for missing key, got %q", got)
	}
}

func TestExtractItemID_NonObjectMidPath_ReturnsError(t *testing.T) {
	item := map[string]interface{}{"meta": "not-an-object"}
	_, err := extractItemID(item, "meta.uid")
	if err == nil {
		t.Fatal("expected error for non-object mid-path segment, got nil")
	}
}

func TestExtractItemID_WhitespaceString_TrimsToEmpty(t *testing.T) {
	got, err := extractItemID(map[string]interface{}{"id": "   "}, "id")
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if got != "" {
		t.Errorf("expected whitespace-only id to trim to empty, got %q", got)
	}
}

// TestBuildConnectorDirectoryInputs_WhitespaceItemID_ReturnsError ensures a
// whitespace-only id fails closed instead of collapsing the directory to the
// scheduler's global reservation sentinel after the scheduler trims the key.
func TestBuildConnectorDirectoryInputs_WhitespaceItemID_ReturnsError(t *testing.T) {
	exec := &FlowExecutor{
		connectorDirectoryReader: &fakeDirectoryReader{items: []map[string]interface{}{{"id": "  "}}},
	}
	if _, err := exec.buildConnectorDirectoryInputs(context.Background(), makeFlowVersionWithSource(validSrc()), nil); err == nil {
		t.Fatal("expected error for whitespace-only itemId, got nil")
	}
}

// ---------------------------------------------------------------------------
// FIX 1 — fail-closed guard: ConnectorSource must not coexist with local-query
// ---------------------------------------------------------------------------

func TestBuildConnectorDirectoryInputs_StartFiltersSet_ReturnsError(t *testing.T) {
	exec := &FlowExecutor{
		connectorDirectoryReader: &fakeDirectoryReader{items: []map[string]interface{}{}},
	}
	fv := &models.FlowVersion{
		DefinitionOwnerPack: "test-pack",
		Metadata: models.FlowMetadata{
			Start: &models.FlowStart{
				ConnectorSource: validSrc(),
				Filters: []models.FlowRecordFilter{
					{Field: "Status", Operator: "EqualTo", Value: "Active"},
				},
			},
		},
	}
	_, err := exec.buildConnectorDirectoryInputs(context.Background(), fv, nil)
	if err == nil {
		t.Fatal("expected error when Start.Filters is set alongside ConnectorSource, got nil")
	}
}

func TestBuildConnectorDirectoryInputs_StartObjectSet_ReturnsError(t *testing.T) {
	exec := &FlowExecutor{
		connectorDirectoryReader: &fakeDirectoryReader{items: []map[string]interface{}{}},
	}
	fv := &models.FlowVersion{
		DefinitionOwnerPack: "test-pack",
		Metadata: models.FlowMetadata{
			Start: &models.FlowStart{
				ConnectorSource: validSrc(),
				Object:          "Contact",
			},
		},
	}
	_, err := exec.buildConnectorDirectoryInputs(context.Background(), fv, nil)
	if err == nil {
		t.Fatal("expected error when Start.Object is set alongside ConnectorSource, got nil")
	}
}

// ---------------------------------------------------------------------------
// json.Number id handling — ensures numeric ids are not coerced to sci-notation
// ---------------------------------------------------------------------------

// TestBuildConnectorDirectoryInputs_NonScalarItemID_ReturnsError confirms that
// extractItemID rejects map and slice leaves — a non-scalar id path indicates
// metadata error / schema drift and must fail closed, not fabricate a synthetic Id.
func TestBuildConnectorDirectoryInputs_NonScalarItemID_ReturnsError(t *testing.T) {
	for _, tc := range []struct {
		name string
		id   interface{}
	}{
		{"map leaf", map[string]interface{}{"nested": "val"}},
		{"slice leaf", []interface{}{"a", "b"}},
	} {
		t.Run(tc.name, func(t *testing.T) {
			items := []map[string]interface{}{
				{"id": tc.id, "name": "Widget"},
			}
			exec := &FlowExecutor{
				connectorDirectoryReader: &fakeDirectoryReader{items: items},
			}
			fv := makeFlowVersionWithSource(validSrc())
			_, err := exec.buildConnectorDirectoryInputs(context.Background(), fv, nil)
			if err == nil {
				t.Fatalf("expected error for non-scalar id (%T), got nil", tc.id)
			}
		})
	}
}

// TestBuildConnectorDirectoryInputs_DuplicateItemID_ReturnsError confirms that
// two items sharing the same id fail closed before any are emitted.
func TestBuildConnectorDirectoryInputs_DuplicateItemID_ReturnsError(t *testing.T) {
	items := []map[string]interface{}{
		{"id": "same-id", "name": "First"},
		{"id": "same-id", "name": "Second"},
	}
	exec := &FlowExecutor{
		connectorDirectoryReader: &fakeDirectoryReader{items: items},
	}
	fv := makeFlowVersionWithSource(validSrc())
	_, err := exec.buildConnectorDirectoryInputs(context.Background(), fv, nil)
	if err == nil {
		t.Fatal("expected error for duplicate item id, got nil")
	}
	if !strings.Contains(err.Error(), "duplicate") {
		t.Errorf("expected 'duplicate' in error, got: %q", err.Error())
	}
}

// TestBuildConnectorDirectoryInputs_BoolItemID_ReturnsError confirms that a
// boolean id is no longer accepted — bool has a 2-value keyspace and is never
// a legitimate unique directory id.
func TestBuildConnectorDirectoryInputs_BoolItemID_ReturnsError(t *testing.T) {
	items := []map[string]interface{}{
		{"id": true, "name": "Widget"},
	}
	exec := &FlowExecutor{
		connectorDirectoryReader: &fakeDirectoryReader{items: items},
	}
	fv := makeFlowVersionWithSource(validSrc())
	_, err := exec.buildConnectorDirectoryInputs(context.Background(), fv, nil)
	if err == nil {
		t.Fatal("expected error for bool item id, got nil")
	}
}

func TestBuildConnectorDirectoryInputs_JsonNumberID_NoSciNotation(t *testing.T) {
	// Simulate what extractDirectoryItems emits when dec.UseNumber() is active:
	// numeric ids arrive as json.Number, not float64.
	numericID := json.Number("1234567")
	items := []map[string]interface{}{
		{"id": numericID, "name": "Widget"},
	}
	exec := &FlowExecutor{
		connectorDirectoryReader: &fakeDirectoryReader{items: items},
	}
	fv := makeFlowVersionWithSource(validSrc())
	sets, err := exec.buildConnectorDirectoryInputs(context.Background(), fv, nil)
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if len(sets) != 1 {
		t.Fatalf("expected 1 input set, got %d", len(sets))
	}
	id, ok := sets[0]["Id"].(string)
	if !ok {
		t.Fatalf("top-level Id is not a string: %T", sets[0]["Id"])
	}
	if id != "1234567" {
		t.Errorf("expected Id=1234567 (no sci-notation), got %q", id)
	}
}

// ---------------------------------------------------------------------------
// ValidateConnectorSource unit tests (shared validator, table-driven)
// ---------------------------------------------------------------------------

func TestValidateConnectorSource_NilStart_ReturnsNil(t *testing.T) {
	if err := models.ValidateConnectorSource(nil); err != nil {
		t.Fatalf("expected nil for nil start, got: %v", err)
	}
}

func TestValidateConnectorSource_NilConnectorSource_ReturnsNil(t *testing.T) {
	start := &models.FlowStart{}
	if err := models.ValidateConnectorSource(start); err != nil {
		t.Fatalf("expected nil when ConnectorSource is nil, got: %v", err)
	}
}

func TestValidateConnectorSource_ValidSource_ReturnsNil(t *testing.T) {
	start := &models.FlowStart{ConnectorSource: validSrc()}
	if err := models.ValidateConnectorSource(start); err != nil {
		t.Fatalf("expected nil for fully valid source, got: %v", err)
	}
}

func TestValidateConnectorSource_RequiredFieldBlank(t *testing.T) {
	cases := []struct {
		name  string
		mutFn func(*models.FlowConnectorSource)
	}{
		{"blank OwnerPack", func(s *models.FlowConnectorSource) { s.OwnerPack = "" }},
		{"whitespace OwnerPack", func(s *models.FlowConnectorSource) { s.OwnerPack = "  " }},
		{"blank ProfileName", func(s *models.FlowConnectorSource) { s.ProfileName = "" }},
		{"blank ObjectKey", func(s *models.FlowConnectorSource) { s.ObjectKey = "" }},
		{"blank ItemsPath", func(s *models.FlowConnectorSource) { s.ItemsPath = "" }},
		{"blank ItemIDPath", func(s *models.FlowConnectorSource) { s.ItemIDPath = "" }},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			src := validSrc()
			tc.mutFn(src)
			start := &models.FlowStart{ConnectorSource: src}
			if err := models.ValidateConnectorSource(start); err == nil {
				t.Fatalf("expected non-nil error for %q, got nil", tc.name)
			}
		})
	}
}

func TestValidateConnectorSource_LocalQueryFieldsCoexist_ReturnsError(t *testing.T) {
	cases := []struct {
		name  string
		start *models.FlowStart
	}{
		{
			"Object set",
			&models.FlowStart{ConnectorSource: validSrc(), Object: "Contact"},
		},
		{
			"FilterLogic set",
			&models.FlowStart{ConnectorSource: validSrc(), FilterLogic: "1 AND 2"},
		},
		{
			"FilterFormula set",
			&models.FlowStart{ConnectorSource: validSrc(), FilterFormula: "Name != null"},
		},
		{
			"Filters set",
			&models.FlowStart{
				ConnectorSource: validSrc(),
				Filters: []models.FlowRecordFilter{
					{Field: "Status", Operator: "EqualTo", Value: "Active"},
				},
			},
		},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			if err := models.ValidateConnectorSource(tc.start); err == nil {
				t.Fatalf("expected non-nil error for %q, got nil", tc.name)
			}
		})
	}
}

// ---------------------------------------------------------------------------
// R6a — deadline test: PullDirectory must receive a context with a deadline
// ---------------------------------------------------------------------------

// deadlineCapturingReader records whether the context passed to PullDirectory
// has a deadline set. Used to assert R6a's per-run timeout wrapping.
type deadlineCapturingReader struct {
	items       []map[string]interface{}
	hadDeadline bool
}

func (d *deadlineCapturingReader) PullDirectory(ctx context.Context, _ ConnectorDirectoryPullRequest) ([]map[string]interface{}, error) {
	_, d.hadDeadline = ctx.Deadline()
	return d.items, nil
}

func TestBuildConnectorDirectoryInputs_PullReceivesDeadline(t *testing.T) {
	reader := &deadlineCapturingReader{
		items: []map[string]interface{}{{"id": "x1"}},
	}
	exec := &FlowExecutor{connectorDirectoryReader: reader}
	fv := makeFlowVersionWithSource(validSrc())
	// Pass a background context (no deadline) so only the builder's wrapping can set one.
	if _, err := exec.buildConnectorDirectoryInputs(context.Background(), fv, nil); err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if !reader.hadDeadline {
		t.Fatal("PullDirectory received a context without a deadline; expected per-run timeout wrapping (R6a)")
	}
}