package plan

import (
	"sort"
	"strings"
	"testing"

	"github.com/stretchr/testify/require"
)

func TestCompileWorkItemFanOutV4(t *testing.T) {
	artifact, diagnostics, err := Compile([]byte(workItemFanOutPlan()))
	require.NoError(t, err)
	require.Empty(t, diagnostics)
	require.NotNil(t, artifact)
	require.Equal(t, ArtifactVersionV4, artifact.ArtifactVersion)
	require.Equal(t, CompilerVersionV4, artifact.CompilerVersion)

	producer := artifactNodeByID(artifact, "plan_work")
	require.NotNil(t, producer.Config.SendMessage.Responder)
	require.Equal(t, ArtifactLoadActivationInput, producer.Config.SendMessage.Responder.Load.Space)
	require.NotNil(t, producer.Config.SendMessage.RelatedScopesArray)
	require.Nil(t, producer.Config.SendMessage.RelatedScopes)
	require.NotNil(t, producer.Config.SendMessage.WorkItems)
	require.Equal(t, ArtifactLoadActivationInput, producer.Config.SendMessage.WorkItems.AllowedResponders.Load.Space)
	workItems, ok := artifactContractFieldSchema(artifactPortContract(producer.OutputPorts, "completed"), sendMessageOutputWorkItems)
	require.True(t, ok)
	require.Equal(t, TypeWorkItems, workItems.Type)
	require.True(t, workItems.Required)

	fanOut := artifactNodeByID(artifact, "fan_out_work")
	require.Equal(t, OpcodeFanOut, fanOut.Opcode)
	require.Equal(t, DefaultFanOutMaxConcurrency, fanOut.Config.FanOut.MaxConcurrency)
	require.Equal(t, FanOutFailureCollect, fanOut.Config.FanOut.FailureMode)
	itemContract, err := lowerArtifactContract(FanOutItemContract())
	require.NoError(t, err)
	require.Equal(t, itemContract, artifactPortContract(fanOut.OutputPorts, "item"))

	implement := artifactNodeByID(artifact, "implement")
	require.Equal(t, []ArtifactScopeFrame{{Kind: ArtifactScopeFanOut, OwnerID: "fan_out_work"}}, implement.ScopePath)
	itemEdge := artifactEdgeByID(t, artifact, "fan_out_to_implement")
	loads := map[string]ArtifactLoad{}
	for _, binding := range itemEdge.Bindings {
		require.NotNil(t, binding.Value.Load)
		loads[binding.Target] = *binding.Value.Load
	}
	require.Equal(t, ArtifactLoadFanOutItem, loads[WorkItemFieldKey].Space)
	require.Equal(t, ArtifactLoadFanOutItem, loads[FanOutItemFieldDependencies].Space)

	join := artifactNodeByID(artifact, "join_work")
	require.Equal(t, "fan_out_work", join.Config.Join.OwnerID)
	require.Len(t, join.Config.Join.Merge, 1)
	require.Equal(t, ArtifactLoadJoinItems, join.Config.Join.Merge[0].Value.Load.Space)
	require.Equal(t, "fan_out_work", join.Config.Join.Merge[0].Value.Load.ScopeOwnerID)

	require.Len(t, artifact.Scopes, 1)
	scope := artifact.Scopes[0]
	require.Equal(t, ArtifactScopeFanOut, scope.Kind)
	require.Equal(t, "fan_out_work", scope.OwnerID)
	require.Equal(t, "join_work", scope.FanOut.JoinID)
	require.Equal(t, []string{"implement"}, scope.FanOut.Members)
	require.Equal(t, []string{"fan_out_to_implement"}, scope.FanOut.BodyEntryEdges)
	require.Equal(t, []string{"implement_to_join"}, scope.FanOut.SuccessArrivals)

	failed := artifactEdgeByID(t, artifact, implicitFailureEdgeID("implement", "failed"))
	require.Contains(t, failed.ScopeTransitions, ArtifactScopeTransition{
		ScopeKind: ArtifactScopeFanOut,
		OwnerID:   "fan_out_work",
		Action:    "fail",
	})

	payload, err := artifact.DeterministicJSON()
	require.NoError(t, err)
	require.NoError(t, ValidateEnvelope(artifact))
	decoded, err := DecodeArtifact(payload)
	require.NoError(t, err)
	require.Equal(t, artifact.Digest, decoded.Digest)
}

func TestFanOutMaxConcurrencyIsBoundedAndFrozen(t *testing.T) {
	source := workItemFanOutPlan()
	for _, test := range []struct {
		name         string
		value        string
		wantCode     string
		wantArtifact int
	}{
		{name: "omitted", wantArtifact: DefaultFanOutMaxConcurrency},
		{name: "explicit", value: "        maxConcurrency: 7\n", wantArtifact: 7},
		{name: "zero", value: "        maxConcurrency: 0\n", wantCode: "fan_out.invalid_max_concurrency"},
		{name: "too large", value: "        maxConcurrency: 65\n", wantCode: "fan_out.max_concurrency_too_large"},
	} {
		t.Run(test.name, func(t *testing.T) {
			candidate := strings.Replace(source, "        failureMode: collect\n", test.value+"        failureMode: collect\n", 1)
			artifact, diagnostics, err := Compile([]byte(candidate))
			require.NoError(t, err)
			if test.wantCode != "" {
				require.Nil(t, artifact)
				require.True(t, hasDiagnosticCode(diagnostics, test.wantCode), diagnostics)
				return
			}
			require.Empty(t, diagnostics)
			require.Equal(t, test.wantArtifact, artifactNodeByID(artifact, "fan_out_work").Config.FanOut.MaxConcurrency)
		})
	}
}

func TestWorkItemContractIsClosedAndNormalized(t *testing.T) {
	contract := WorkItemContract()
	require.Len(t, contract, 6)
	require.ElementsMatch(t, []string{
		WorkItemFieldKey,
		WorkItemFieldDependsOn,
		WorkItemFieldWorkScope,
		WorkItemFieldRelatedScopes,
		WorkItemFieldResponder,
		WorkItemFieldInstruction,
	}, sortedSchemaKeys(contract))

	dependsOn := contract[WorkItemFieldDependsOn]
	require.False(t, schemaRequired(dependsOn))
	require.Equal(t, []any{}, dependsOn.Default)
	require.NotNil(t, dependsOn.Items)
	require.Equal(t, TypeString, dependsOn.Items.Type)
	require.True(t, schemaRequired(contract[WorkItemFieldKey]))
	require.True(t, schemaRequired(contract[WorkItemFieldRelatedScopes]))

	item := FanOutItemContract()
	require.Len(t, item, 7)
	require.NotContains(t, contract, FanOutItemFieldDependencies)
	require.Equal(t, TypeObject, item[FanOutItemFieldDependencies].Type)
	require.True(t, schemaRequired(item[FanOutItemFieldDependencies]))
}

func TestWorkItemsSchemaIsReservedForCompilerOwnedTransport(t *testing.T) {
	tests := []struct {
		name   string
		source func(string) string
	}{
		{
			name: "plan input",
			source: func(source string) string {
				return strings.Replace(source, "input:\n", "input:\n  forged_items: {type: work_items, required: true}\n", 1)
			},
		},
		{
			name: "plan output",
			source: func(source string) string {
				return strings.Replace(source, "output:\n", "output:\n  forged_items: {type: work_items, required: true}\n", 1)
			},
		},
		{
			name: "plan state",
			source: func(source string) string {
				return strings.Replace(source, "graph:\n", "state:\n  forged_items: {type: work_items, required: true}\ngraph:\n", 1)
			},
		},
		{
			name: "callback output alias",
			source: func(source string) string {
				return strings.Replace(
					source,
					"        callbackOutput:\n          summary: {type: string, required: true}\n",
					"        callbackOutput:\n          forged_items: {type: work_items, required: true}\n          summary: {type: string, required: true}\n",
					1,
				)
			},
		},
		{
			name: "intermediate node input",
			source: func(source string) string {
				return strings.Replace(
					source,
					"    - id: implement\n      type: send_message\n      input:\n",
					"    - id: implement\n      type: send_message\n      input:\n        forged_items: {type: work_items, required: true}\n",
					1,
				)
			},
		},
		{
			name: "join result",
			source: func(source string) string {
				return strings.Replace(
					source,
					"        result:\n          results:\n",
					"        result:\n          forged_items: {type: work_items, required: true}\n          results:\n",
					1,
				)
			},
		},
		{
			name: "loop state",
			source: func(source string) string {
				return strings.Replace(
					source,
					"    - id: done\n",
					"    - id: illegal_loop\n"+
						"      type: loop\n"+
						"      loop:\n"+
						"        mode: counted\n"+
						"        iterations: 1\n"+
						"        state:\n"+
						"          forged_items: {type: work_items, required: true}\n"+
						"    - id: done\n",
					1,
				)
			},
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			artifact, diagnostics, err := Compile([]byte(test.source(workItemFanOutPlan())))
			require.NoError(t, err)
			require.Nil(t, artifact)
			require.True(t, hasDiagnosticCode(diagnostics, "schema.work_items_reserved"), diagnostics)
		})
	}
}

func TestFanOutWorkItemsInputCannotDeclareAnAuthoredDefault(t *testing.T) {
	source := strings.Replace(
		workItemFanOutPlan(),
		"items: {type: work_items, required: true}",
		"items: {type: work_items, required: true, default: []}",
		1,
	)
	artifact, diagnostics, err := Compile([]byte(source))
	require.NoError(t, err)
	require.Nil(t, artifact)
	require.True(t, hasDiagnosticCode(diagnostics, "schema.default_type"), diagnostics)
}

func TestFanOutRejectsAuthoredWorkItemsProvenanceBypasses(t *testing.T) {
	t.Run("callback output alias bypasses responder allowlist", func(t *testing.T) {
		source := strings.Replace(
			workItemFanOutPlan(),
			"        callbackOutput:\n          summary: {type: string, required: true}\n",
			"        callbackOutput:\n          forged_items: {type: work_items, required: true}\n          summary: {type: string, required: true}\n",
			1,
		)
		source = strings.Replace(source, "items: {ref: output.workItems}", "items: {ref: output.forged_items}", 1)

		artifact, diagnostics, err := Compile([]byte(source))
		require.NoError(t, err)
		require.Nil(t, artifact)
		require.True(t, hasDiagnosticCode(diagnostics, "schema.work_items_reserved"), diagnostics)
		require.True(t, hasDiagnosticCode(diagnostics, "fan_out.items_source"), diagnostics)
	})

	t.Run("plan input bypasses callback scope authorization", func(t *testing.T) {
		source := strings.Replace(
			workItemFanOutPlan(),
			"input:\n",
			"input:\n  forged_items: {type: work_items, required: true}\n",
			1,
		)
		source = strings.Replace(source, "items: {ref: input.items}", "items: {ref: plan.input.forged_items}", 1)

		artifact, diagnostics, err := Compile([]byte(source))
		require.NoError(t, err)
		require.Nil(t, artifact)
		require.True(t, hasDiagnosticCode(diagnostics, "schema.work_items_reserved"), diagnostics)
		require.True(t, hasDiagnosticCode(diagnostics, "fan_out.items_source"), diagnostics)
	})
}

func TestWorkItemsCallbackRequiresOneDirectFanOutConsumer(t *testing.T) {
	t.Run("missing consumer", func(t *testing.T) {
		source := strings.Replace(
			workItemFanOutPlan(),
			"        items: {ref: output.workItems}\n",
			"        items: {value: []}\n",
			1,
		)
		artifact, diagnostics, err := Compile([]byte(source))
		require.NoError(t, err)
		require.Nil(t, artifact)
		require.True(t, hasDiagnosticCode(diagnostics, "send_message.work_items_consumer"), diagnostics)
	})

	t.Run("multiple consumers", func(t *testing.T) {
		const edge = "    - id: plan_to_fan_out\n" +
			"      source: {nodeId: plan_work, port: completed}\n" +
			"      target: {nodeId: fan_out_work, port: in}\n" +
			"      bindings:\n" +
			"        items: {ref: output.workItems}\n"
		source := strings.Replace(
			workItemFanOutPlan(),
			edge,
			edge+strings.Replace(edge, "plan_to_fan_out", "plan_to_fan_out_duplicate", 1),
			1,
		)
		artifact, diagnostics, err := Compile([]byte(source))
		require.NoError(t, err)
		require.Nil(t, artifact)
		require.True(t, hasDiagnosticCode(diagnostics, "send_message.work_items_consumer"), diagnostics)
	})
}

func TestWorkItemsDependencyValueContractExtractionIsUniqueAndDetached(t *testing.T) {
	artifact := compileArtifactForTest(t, workItemFanOutPlan())
	join := artifactNodeByID(artifact, "join_work")
	for portIndex := range join.InputPorts {
		if join.InputPorts[portIndex].Name != "item" {
			continue
		}
		join.InputPorts[portIndex].Contract[0].Schema.Type = TypeArray
		join.InputPorts[portIndex].Contract[0].Schema.Items = &ArtifactSchema{
			Type: TypeNumber, Required: true, Values: []string{},
		}
	}

	contract, err := WorkItemsDependencyValueContract(artifact, "plan_work")
	require.NoError(t, err)
	require.Len(t, contract, 1)
	require.Equal(t, "result", contract[0].Name)
	require.Equal(t, TypeArray, contract[0].Schema.Type)
	require.NotNil(t, contract[0].Schema.Items)
	require.Equal(t, TypeNumber, contract[0].Schema.Items.Type)

	contract[0].Name = "mutated"
	contract[0].Schema.Items.Type = TypeString
	again, err := WorkItemsDependencyValueContract(artifact, "plan_work")
	require.NoError(t, err)
	require.Equal(t, "result", again[0].Name)
	require.Equal(t, TypeNumber, again[0].Schema.Items.Type)

	duplicate := artifactEdgeByID(t, artifact, "plan_to_fan_out")
	cloned := *duplicate
	cloned.ID = "plan_to_fan_out_duplicate"
	cloned.Bindings = append([]ArtifactBinding{}, duplicate.Bindings...)
	artifact.Edges = append(artifact.Edges, cloned)
	sort.Slice(artifact.Edges, func(i, j int) bool { return artifact.Edges[i].ID < artifact.Edges[j].ID })
	_, err = WorkItemsDependencyValueContract(artifact, "plan_work")
	require.ErrorContains(t, err, "must feed exactly one direct fan_out")
}

func TestArtifactEnvelopeRejectsAuthoredWorkItemsSmuggling(t *testing.T) {
	t.Run("callback output alias", func(t *testing.T) {
		artifact := compileArtifactForTest(t, workItemFanOutPlan())
		producer := artifactNodeByID(artifact, "plan_work")
		forged := ArtifactField{
			Name: "forged_items",
			Schema: ArtifactSchema{
				Type:     TypeWorkItems,
				Required: true,
				Values:   []string{},
			},
		}
		producer.Config.SendMessage.CallbackOutput = append(producer.Config.SendMessage.CallbackOutput, forged)
		sort.Slice(producer.Config.SendMessage.CallbackOutput, func(i, j int) bool {
			return producer.Config.SendMessage.CallbackOutput[i].Name < producer.Config.SendMessage.CallbackOutput[j].Name
		})
		for portIndex := range producer.OutputPorts {
			if producer.OutputPorts[portIndex].Name != "completed" {
				continue
			}
			producer.OutputPorts[portIndex].Contract = append(producer.OutputPorts[portIndex].Contract, forged)
			sort.Slice(producer.OutputPorts[portIndex].Contract, func(i, j int) bool {
				return producer.OutputPorts[portIndex].Contract[i].Name < producer.OutputPorts[portIndex].Contract[j].Name
			})
		}
		edge := artifactEdgeByID(t, artifact, "plan_to_fan_out")
		for bindingIndex := range edge.Bindings {
			if edge.Bindings[bindingIndex].Target == "items" {
				edge.Bindings[bindingIndex].Value.Load.Path = []string{"forged_items"}
			}
		}
		refreshArtifactDigest(t, artifact)

		require.ErrorContains(t, ValidateEnvelope(artifact), "cannot declare work_items")
	})

	t.Run("plan input", func(t *testing.T) {
		artifact := compileArtifactForTest(t, workItemFanOutPlan())
		artifact.Contracts.Input = append(artifact.Contracts.Input, ArtifactField{
			Name: "forged_items",
			Schema: ArtifactSchema{
				Type:     TypeWorkItems,
				Required: true,
				Values:   []string{},
			},
		})
		sort.Slice(artifact.Contracts.Input, func(i, j int) bool {
			return artifact.Contracts.Input[i].Name < artifact.Contracts.Input[j].Name
		})
		refreshArtifactDigest(t, artifact)

		require.ErrorContains(t, ValidateEnvelope(artifact), "contracts.input cannot declare work_items")
	})
}

func refreshArtifactDigest(t *testing.T, artifact *Artifact) {
	t.Helper()
	artifact.Digest = ""
	digest, err := artifact.computeDigest()
	require.NoError(t, err)
	artifact.Digest = digest
}

func TestArtifactV4RequiresExactCompilerVersionAndRejectsV4AsV3(t *testing.T) {
	artifact := compileArtifactForTest(t, workItemFanOutPlan())

	wrongCompiler := *artifact
	wrongCompiler.CompilerVersion = CompilerVersionV3
	wrongCompiler.Digest = ""
	digest, err := wrongCompiler.computeDigest()
	require.NoError(t, err)
	wrongCompiler.Digest = digest
	require.ErrorContains(t, ValidateEnvelope(&wrongCompiler), "requires compilerVersion")

	downgraded := *artifact
	downgraded.ArtifactVersion = ArtifactVersionV3
	downgraded.CompilerVersion = CompilerVersionV3
	downgraded.Digest = ""
	digest, err = downgraded.computeDigest()
	require.NoError(t, err)
	downgraded.Digest = digest
	require.ErrorContains(t, ValidateEnvelope(&downgraded), "does not support v4 Workflow features")

	legacy := compileArtifactForTest(t, validCallbackPlan)
	legacy.ArtifactVersion = ArtifactVersionV4
	legacy.CompilerVersion = CompilerVersionV4
	legacy.Digest = ""
	digest, err = legacy.computeDigest()
	require.NoError(t, err)
	legacy.Digest = digest
	require.ErrorContains(t, ValidateEnvelope(legacy), "requires at least one v4 Workflow feature")

	legacyWithV4InputVocabulary := compileArtifactForTest(t, validCallbackPlan)
	notify := artifactNodeByID(legacyWithV4InputVocabulary, "notify")
	notify.InputPorts[0].Contract = append(notify.InputPorts[0].Contract, ArtifactField{
		Name: WorkItemFieldWorkScope,
		Schema: ArtifactSchema{
			Type: TypeString, Values: []string{},
		},
	})
	legacyWithV4InputVocabulary.Digest = ""
	digest, err = legacyWithV4InputVocabulary.computeDigest()
	require.NoError(t, err)
	legacyWithV4InputVocabulary.Digest = digest
	require.ErrorContains(t, ValidateEnvelope(legacyWithV4InputVocabulary), "does not support v4 Workflow features")

	legacyWire, err := compileArtifactForTest(t, validCallbackPlan).DeterministicJSON()
	require.NoError(t, err)
	withNullFanOut := strings.Replace(
		string(legacyWire),
		`"config":{"sendMessage":`,
		`"config":{"fanOut":null,"sendMessage":`,
		1,
	)
	_, err = DecodeArtifact([]byte(withNullFanOut))
	require.ErrorContains(t, err, "does not support v4 Workflow features")
}

func TestSelectedResponderAloneRequiresArtifactV4(t *testing.T) {
	artifact := compileArtifactForTest(t, `apiVersion: loop.plan/v1
kind: PlanVersion
input:
  destination: {type: channel, required: true}
  responder: {type: agent, required: true}
graph:
  entry:
    nodeId: ask
    port: in
    input:
      destination: {ref: plan.input.destination}
      responder: {ref: plan.input.responder}
  nodes:
  - id: ask
    type: send_message
    input:
      destination: {type: channel, required: true}
      responder: {type: agent, required: true}
    sendMessage:
      target: "{{ input.destination }}"
      responder: {ref: input.responder}
      text: Review this result.
      completion: callback
      callbackOutput:
        result: {type: string, required: true}
  - id: done
    type: end
    end: {outcome: succeeded}
  edges:
  - id: reviewed
    source: {nodeId: ask, port: completed}
    target: {nodeId: done, port: in}
`)
	require.Equal(t, ArtifactVersionV4, artifact.ArtifactVersion)
	require.Equal(t, CompilerVersionV4, artifact.CompilerVersion)
	require.NotNil(t, artifactNodeByID(artifact, "ask").Config.SendMessage.Responder)
}

func TestJoinFanOutAndForkOwnersAreMutuallyExclusive(t *testing.T) {
	source := strings.Replace(
		workItemFanOutPlan(),
		"        fanOutId: fan_out_work\n",
		"        forkId: fan_out_work\n        fanOutId: fan_out_work\n",
		1,
	)
	artifact, diagnostics, err := Compile([]byte(source))
	require.NoError(t, err)
	require.Nil(t, artifact)
	require.True(t, hasDiagnosticCode(diagnostics, "join.owner_required"), diagnostics)
}

func TestRelatedScopesArrayExpressionRemainsStrict(t *testing.T) {
	source := strings.Replace(
		workItemFanOutPlan(),
		"        relatedScopes: {ref: input.requirement_scopes}\n",
		"        relatedScopes: {ref: input.requirement_scopes, typo: true}\n",
		1,
	)
	artifact, diagnostics, err := Compile([]byte(source))
	require.NoError(t, err)
	require.Nil(t, artifact)
	require.True(t, hasDiagnosticCode(diagnostics, "syntax.unknown_field"), diagnostics)
}

func TestFanOutScopeRejectsAnOpenBody(t *testing.T) {
	source := strings.Replace(
		workItemFanOutPlan(),
		"      target: {nodeId: join_work, port: item}\n",
		"      target: {nodeId: done, port: in}\n",
		1,
	)
	artifact, diagnostics, err := Compile([]byte(source))
	require.NoError(t, err)
	require.Nil(t, artifact)
	require.True(t, hasDiagnosticCode(diagnostics, "fan_out.open_body"), diagnostics)
}

func TestFanOutRejectsUnsupportedModes(t *testing.T) {
	for _, test := range []struct {
		name     string
		old      string
		new      string
		wantCode string
	}{
		{
			name:     "failure mode",
			old:      "        failureMode: collect\n",
			new:      "        failureMode: fail_fast\n",
			wantCode: "fan_out.invalid_failure_mode",
		},
		{
			name:     "join mode",
			old:      "        mode: all\n",
			new:      "        mode: any\n",
			wantCode: "join.fan_out_mode",
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			source := strings.Replace(workItemFanOutPlan(), test.old, test.new, 1)
			artifact, diagnostics, err := Compile([]byte(source))
			require.NoError(t, err)
			require.Nil(t, artifact)
			require.True(t, hasDiagnosticCode(diagnostics, test.wantCode), diagnostics)
		})
	}
}

func TestFanOutRejectsNestedFanOut(t *testing.T) {
	source := strings.Replace(
		workItemFanOutPlan(),
		"  parent_task: {type: scope, required: true}\n",
		"  nested_items: {type: work_items, required: true}\n  parent_task: {type: scope, required: true}\n",
		1,
	)
	source = strings.Replace(
		source,
		"    - id: implement\n",
		"    - id: nested_fan_out\n"+
			"      type: fan_out\n"+
			"      input: {}\n"+
			"      fanOut:\n"+
			"        items: {ref: plan.input.nested_items}\n"+
			"        failureMode: collect\n"+
			"    - id: implement\n",
		1,
	)
	source = strings.Replace(
		source,
		"      target: {nodeId: implement, port: in}\n",
		"      target: {nodeId: nested_fan_out, port: in}\n",
		1,
	)
	source = strings.Replace(
		source,
		"    - id: implement_to_join\n",
		"    - id: nested_to_implement\n"+
			"      source: {nodeId: nested_fan_out, port: item}\n"+
			"      target: {nodeId: implement, port: in}\n"+
			"    - id: implement_to_join\n",
		1,
	)

	artifact, diagnostics, err := Compile([]byte(source))
	require.NoError(t, err)
	require.Nil(t, artifact)
	require.True(t, hasDiagnosticCode(diagnostics, "fan_out.nested_not_allowed"), diagnostics)
}

func TestArtifactRejectsNestedFanOut(t *testing.T) {
	artifact := compileArtifactForTest(t, workItemFanOutPlan())
	artifactNodeByID(artifact, "fan_out_work").ScopePath = []ArtifactScopeFrame{{
		Kind:    ArtifactScopeFanOut,
		OwnerID: "outer_fan_out",
	}}
	artifact.Digest = ""
	digest, err := artifact.computeDigest()
	require.NoError(t, err)
	artifact.Digest = digest
	require.ErrorContains(t, ValidateEnvelope(artifact), "nested fan_out")
}

func workItemFanOutPlan() string {
	return `apiVersion: loop.plan/v1
kind: PlanVersion
input:
  planner: {type: agent, required: true}
  developers:
    type: array
    items: {type: agent}
    required: true
  parent_task: {type: scope, required: true}
  requirement_scopes:
    type: array
    items: {type: scope}
    required: true
output:
  results:
    type: array
    items: {type: object}
    required: true
graph:
  entry:
    nodeId: plan_work
    port: in
  nodes:
    - id: plan_work
      type: send_message
      input:
        planner: {type: agent, required: true}
        developers:
          type: array
          items: {type: agent}
          required: true
        parent_task: {type: scope, required: true}
        requirement_scopes:
          type: array
          items: {type: scope}
          required: true
      sendMessage:
        scope: {ref: input.parent_task}
        responder: {ref: input.planner}
        text: Decompose this work
        completion: callback
        workScope: {ref: input.parent_task}
        relatedScopes: {ref: input.requirement_scopes}
        workItems:
          allowedResponders: {ref: input.developers}
        callbackOutput:
          summary: {type: string, required: true}
    - id: fan_out_work
      type: fan_out
      input:
        items: {type: work_items, required: true}
      fanOut:
        items: {ref: input.items}
        failureMode: collect
    - id: implement
      type: send_message
      input:
        key: {type: string, required: true}
        dependsOn:
          type: array
          items: {type: string}
          default: []
        workScope: {type: scope, required: true}
        relatedScopes:
          type: array
          items: {type: scope}
          required: true
        responder: {type: responder, required: true}
        instruction: {type: string, required: true}
        dependencies: {type: object, required: true}
      sendMessage:
        scope: {ref: input.workScope}
        responder: {ref: input.responder}
        text: "{{ input.instruction }}"
        completion: callback
        workScope: {ref: input.workScope}
        relatedScopes: {ref: input.relatedScopes}
        callbackOutput:
          result: {type: object, required: true}
    - id: join_work
      type: join
      input:
        result: {type: object, required: true}
      join:
        fanOutId: fan_out_work
        mode: all
        result:
          results:
            type: array
            items: {type: object}
            required: true
        merge:
          results: {ref: items}
    - id: done
      type: end
      end: {outcome: succeeded}
  edges:
    - id: plan_to_fan_out
      source: {nodeId: plan_work, port: completed}
      target: {nodeId: fan_out_work, port: in}
      bindings:
        items: {ref: output.workItems}
    - id: fan_out_to_implement
      source: {nodeId: fan_out_work, port: item}
      target: {nodeId: implement, port: in}
    - id: implement_to_join
      source: {nodeId: implement, port: completed}
      target: {nodeId: join_work, port: item}
      bindings:
        result: {ref: output.result}
    - id: join_to_done
      source: {nodeId: join_work, port: completed}
      target: {nodeId: done, port: in}
`
}