package run

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

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/workflow/invocation"
	"github.com/tidbcloud/AgentTeam/internal/workflow/plan"
	"github.com/tidbcloud/AgentTeam/internal/workflow/runlimit"
)

func TestFanOutSixtyFourRootOutputsFailOnlyOverBudgetItems(t *testing.T) {
	t.Parallel()
	kernel := compilePlanKernel(t, strings.Replace(
		fanOutRuntimePlan,
		"maxConcurrency: 2",
		"maxConcurrency: 64",
		1,
	))
	items := make([]any, 64)
	for index := 0; index < len(items)-1; index++ {
		items[index] = workItemRaw(fmt.Sprintf("large_%02d", index), nil, "@dev1")
	}
	items[len(items)-1] = workItemRaw("independent", nil, "@dev2")
	state := startFanOut(t, kernel, items)
	require.Equal(t, 64, countWorkItemStatus(onlyFanOutScope(t, state), WorkItemRunning))

	intents := make([]Intent, 0, len(items))
	for range items {
		var intent Intent
		intent, state = stepToMessageIntent(t, kernel, state)
		intents = append(intents, intent)
	}
	largeAnswer := strings.Repeat("x", 520<<10)
	for _, intent := range intents[:len(intents)-1] {
		state = completeIntent(t, kernel, state, intent, "completed", Values{"answer": largeAnswer}, nil)
	}
	state = completeIntent(
		t,
		kernel,
		state,
		intents[len(intents)-1],
		"completed",
		Values{"answer": "independent result"},
		nil,
	)

	scope := onlyFanOutScope(t, state)
	require.Equal(t, ScopeSucceeded, scope.Status, "collect settles after local resource failures")
	independent := fanOutItemByKey(scope.FanOut, "independent")
	require.NotNil(t, independent)
	require.Equal(t, WorkItemSucceeded, independent.Status, "a small independent item continues")
	resourceFailures := 0
	for index := range scope.FanOut.Items {
		item := &scope.FanOut.Items[index]
		if item.Failure != nil && item.Failure.Code == workItemResourceLimitFailureCode {
			resourceFailures++
			require.Nil(t, item.Value)
		}
	}
	require.Positive(t, resourceFailures, "the admission preflight must reject outputs before activation.Output can grow state without bound")
	encoded, err := json.Marshal(state)
	require.NoError(t, err)
	require.LessOrEqual(t, len(encoded), runlimit.MaxRunStateBytes)

	restored, err := JSONRoundTrip(state)
	require.NoError(t, err)
	require.Equal(t, state, restored)
}

func TestFanOutResourceRejectedCompletionReplayIsIdempotent(t *testing.T) {
	t.Parallel()
	kernel := compilePlanKernel(t, fanOutRuntimePlan)
	state := startFanOut(t, kernel, []any{workItemRaw("root", nil, "@dev1")})
	intent, state := stepToMessageIntent(t, kernel, state)
	completion := Completion{
		ActivationID: intent.ActivationID,
		Port:         "completed",
		Output: Values{
			"answer":    strings.Repeat("x", runlimit.MaxRunPayloadBytes),
			"workScope": messageScopeObject(*intent.WorkScope),
		},
	}

	rejected, err := kernel.Complete(state, completion, reducerNow)
	require.NoError(t, err)
	scope := onlyFanOutScope(t, rejected.State)
	item := fanOutItemByKey(scope.FanOut, "root")
	require.Equal(t, WorkItemFailed, item.Status)
	require.Equal(t, workItemResourceLimitFailureCode, item.Failure.Code)
	activation, err := rejected.State.activation(intent.ActivationID)
	require.NoError(t, err)
	require.Equal(t, ActivationFailed, activation.Status)
	require.NotNil(t, activation.ResourceRejectedCompletion)
	require.Nil(t, activation.Output, "rejected output bytes must not enter durable State")
	stateJSON, err := json.Marshal(rejected.State)
	require.NoError(t, err)
	_, err = invocation.DeterministicJSONObjectWithBudget(
		"workflow Run state",
		stateJSON,
		runlimit.MaxRunStateBytes,
		runlimit.MaxRunStateJSONNesting,
		runlimit.MaxRunStateJSONValues,
	)
	require.NoError(t, err)

	replayed, err := kernel.Complete(rejected.State, completion, reducerNow)
	require.NoError(t, err)
	require.Equal(t, "completion_replayed", replayed.Events[0].Kind)
	require.Equal(t, rejected.State, replayed.State)

	conflicting := completion
	conflicting.Output = cloneValues(completion.Output)
	conflicting.Output["answer"] = strings.Repeat("y", runlimit.MaxRunPayloadBytes)
	_, err = kernel.Complete(rejected.State, conflicting, reducerNow)
	require.ErrorContains(t, err, "already rejected a different completion result")
}

func TestFanOutClosingJoinDoubleBindingFailsOnlyContributingItem(t *testing.T) {
	t.Parallel()
	planSource := strings.Replace(
		fanOutRuntimePlan,
		`      result:
        results:
          type: array
          required: true
          items: {type: object, required: true}
      merge:
        results: {ref: items}`,
		`      result:
        results:
          type: array
          required: true
          items: {type: object, required: true}
        results_copy:
          type: array
          required: true
          items: {type: object, required: true}
      merge:
        results: {ref: items}
        results_copy: {ref: items}`,
		1,
	)
	require.NotEqual(t, fanOutRuntimePlan, planSource)
	kernel := compilePlanKernel(t, planSource)
	state := startFanOut(t, kernel, []any{workItemRaw("root", nil, "@dev1")})
	intent, state := stepToMessageIntent(t, kernel, state)

	completed, err := kernel.Complete(state, Completion{
		ActivationID: intent.ActivationID,
		Port:         "completed",
		Output: Values{
			"answer":    strings.Repeat("x", 2200<<10),
			"workScope": messageScopeObject(*intent.WorkScope),
		},
	}, reducerNow)
	require.NoError(t, err, "closing join resource rejection must not escape as a whole-Run reducer error")
	scope := onlyFanOutScope(t, completed.State)
	item := fanOutItemByKey(scope.FanOut, "root")
	require.Equal(t, WorkItemFailed, item.Status)
	require.Equal(t, workItemResourceLimitFailureCode, item.Failure.Code)
	activation, err := completed.State.activation(intent.ActivationID)
	require.NoError(t, err)
	require.Equal(t, ActivationFailed, activation.Status)
	require.Nil(t, activation.Output)

	terminal := stepUntilTerminal(t, kernel, completed.State, 4)
	require.Equal(t, StatusSucceeded, terminal.Status)
	results := terminal.Output["results"].([]any)
	require.Len(t, results, 1)
	require.Equal(t, string(WorkItemFailed), results[0].(map[string]any)["status"])
	stateJSON, err := json.Marshal(terminal)
	require.NoError(t, err)
	_, err = invocation.DeterministicJSONObjectWithBudget(
		"workflow Run state", stateJSON,
		runlimit.MaxRunStateBytes,
		runlimit.MaxRunStateJSONNesting,
		runlimit.MaxRunStateJSONValues,
	)
	require.NoError(t, err)
}

func TestFanOutStaticOversizeClosingMergeTakesRecoverableFailedPort(t *testing.T) {
	t.Parallel()
	kernel := compilePlanKernel(t, fanOutRepeatedStaticJoinRuntimePlan(17))
	workItems := make([]any, 8)
	for index := range workItems {
		item := workItemRaw(fmt.Sprintf("item_%02d", index), nil, "@dev1")
		item[plan.WorkItemFieldInstruction] = strings.Repeat("i", maxMaterializedWorkItemInstructionBytes)
		workItems[index] = item
	}
	state := startFanOut(t, kernel, workItems)
	for completed := 0; completed < len(workItems); completed++ {
		intent, next := stepToMessageIntent(t, kernel, state)
		state = completeIntent(t, kernel, next, intent, "completed", Values{"answer": "ok"}, nil)
	}
	scope := onlyFanOutScope(t, state)
	require.Equal(t, ScopeFailed, scope.Status)
	for index := range scope.FanOut.Items {
		require.Equal(t, WorkItemSucceeded, scope.FanOut.Items[index].Status)
	}

	terminal := stepUntilTerminal(t, kernel, state, 4)
	require.Equal(t, StatusFailed, terminal.Status)
	join, err := terminal.activation(scope.FanOut.JoinActivationID)
	require.NoError(t, err)
	require.Equal(t, workItemResourceLimitFailureCode, join.Failure.Code)
	require.NoError(t, validateFanOutHardStateBudget("static oversize closing join", terminal))
}

func TestFanOutOrdinaryPureStepRoutingFailsItemBeforeStateStoreBoundary(t *testing.T) {
	t.Parallel()
	kernel := compilePlanKernel(t, fanOutSwitchChainRuntimePlan(12))
	state := startFanOut(t, kernel, []any{workItemRaw("root", nil, "@dev1")})
	intent, state := stepToMessageIntent(t, kernel, state)
	completion := Completion{
		ActivationID: intent.ActivationID,
		Port:         "completed",
		Output: Values{
			"answer":    strings.Repeat("x", 3<<20),
			"workScope": messageScopeObject(*intent.WorkScope),
		},
	}
	completed, err := kernel.Complete(state, completion, reducerNow)
	require.NoError(t, err)
	state = completed.State

	for step := 0; step < 20; step++ {
		scope := onlyFanOutScope(t, state)
		if item := fanOutItemByKey(scope.FanOut, "root"); item.Status.Terminal() {
			break
		}
		transition, stepErr := kernel.Step(state, reducerNow)
		require.NoError(t, stepErr, "ordinary pure fan-out body routing must fail locally")
		state = transition.State
	}
	scope := onlyFanOutScope(t, state)
	item := fanOutItemByKey(scope.FanOut, "root")
	require.Equal(t, WorkItemFailed, item.Status)
	require.Equal(t, workItemResourceLimitFailureCode, item.Failure.Code)
	replayed, err := kernel.Complete(state, completion, reducerNow)
	require.NoError(t, err)
	require.Equal(t, "completion_replayed", replayed.Events[0].Kind)
	require.Equal(t, state, replayed.State)
	encoded, err := json.Marshal(state)
	require.NoError(t, err)
	_, err = invocation.DeterministicJSONObjectWithBudget(
		"workflow Run state", encoded,
		runlimit.MaxRunStateBytes,
		runlimit.MaxRunStateJSONNesting,
		runlimit.MaxRunStateJSONValues,
	)
	require.NoError(t, err)
}

func TestFanOutConcurrentLargeItemsCompactPureHistoryBeforeClosing(t *testing.T) {
	t.Parallel()
	kernel := compilePlanKernel(t, fanOutSwitchChainRuntimePlan(5))
	state := startFanOut(t, kernel, []any{
		workItemRaw("a", nil, "@dev1"),
		workItemRaw("b", nil, "@dev1"),
	})
	aIntent, state := stepToMessageIntent(t, kernel, state)
	bIntent, state := stepToMessageIntent(t, kernel, state)
	state = completeIntent(t, kernel, state, aIntent, "completed", Values{
		"answer": strings.Repeat("a", 1_677_200),
	}, nil)
	state = completeIntent(t, kernel, state, bIntent, "completed", Values{
		"answer": strings.Repeat("b", 1_677_200),
	}, nil)

	for step := 0; step < 16 && !state.Status.Terminal(); step++ {
		transition, err := kernel.Step(state, reducerNow)
		require.NoError(t, err)
		state = transition.State
		require.NoError(t, validateFanOutHardStateBudget("concurrent large item Step", state))
	}
	scope := onlyFanOutScope(t, state)
	require.Equal(t, WorkItemSucceeded, fanOutItemByKey(scope.FanOut, "a").Status)
	require.Equal(t, WorkItemSucceeded, fanOutItemByKey(scope.FanOut, "b").Status)
	require.Equal(t, StatusSucceeded, state.Status)
}

func TestFanOutRejectedCompletionPreservesPriorTerminalContributor(t *testing.T) {
	t.Parallel()
	planSource := strings.Replace(
		fanOutSwitchChainRuntimePlan(5),
		"maxConcurrency: 2",
		"maxConcurrency: 3",
		1,
	)
	kernel := compilePlanKernel(t, planSource)
	state := startFanOut(t, kernel, []any{
		workItemRaw("a", nil, "@dev1"),
		workItemRaw("b", nil, "@dev1"),
		workItemRaw("c", nil, "@dev1"),
	})
	aIntent, state := stepToMessageIntent(t, kernel, state)
	bIntent, state := stepToMessageIntent(t, kernel, state)
	_, state = stepToMessageIntent(t, kernel, state)

	state = completeIntent(
		t,
		kernel,
		state,
		aIntent,
		"completed",
		Values{"answer": strings.Repeat("x", 2_795_540)},
		nil,
	)
	for index := 0; index < 5; index++ {
		transition, err := kernel.Step(state, reducerNow)
		require.NoError(t, err)
		state = transition.State
	}
	require.Equal(t, WorkItemSucceeded, fanOutItemByKey(onlyFanOutScope(t, state).FanOut, "a").Status)

	rejected, err := kernel.Complete(state, Completion{
		ActivationID: bIntent.ActivationID,
		Port:         "completed",
		Output: Values{
			"answer":    strings.Repeat("z", runlimit.MaxRunPayloadBytes),
			"workScope": messageScopeObject(*bIntent.WorkScope),
		},
	}, reducerNow)
	require.NoError(t, err, "actual rejection metadata must settle without escaping the reducer")
	scope := onlyFanOutScope(t, rejected.State)
	require.Equal(t, WorkItemFailed, fanOutItemByKey(scope.FanOut, "b").Status)
	require.Equal(t, workItemResourceLimitFailureCode, fanOutItemByKey(scope.FanOut, "b").Failure.Code)
	require.Equal(t, WorkItemRunning, fanOutItemByKey(scope.FanOut, "c").Status)
	require.Equal(t, WorkItemSucceeded, fanOutItemByKey(scope.FanOut, "a").Status)
	require.NoError(t, validateFanOutHardStateBudget("rejected completion settlement", rejected.State))
}

func TestFanOutClosingPreflightPreservesPriorContributor(t *testing.T) {
	t.Parallel()
	kernel := compilePlanKernel(t, fanOutSwitchChainRuntimePlan(5))
	state := startFanOut(t, kernel, []any{
		workItemRaw("a", nil, "@dev1"),
		workItemRaw("b", nil, "@dev1"),
	})
	aIntent, state := stepToMessageIntent(t, kernel, state)
	bIntent, state := stepToMessageIntent(t, kernel, state)
	state = completeIntent(
		t,
		kernel,
		state,
		aIntent,
		"completed",
		Values{"answer": strings.Repeat("x", 2_600_000)},
		nil,
	)
	for index := 0; index < 5; index++ {
		transition, err := kernel.Step(state, reducerNow)
		require.NoError(t, err)
		state = transition.State
	}
	require.Equal(t, WorkItemSucceeded, fanOutItemByKey(onlyFanOutScope(t, state).FanOut, "a").Status)

	state = completeIntent(t, kernel, state, bIntent, "completed", Values{"answer": "b"}, nil)
	for index := 0; index < 5; index++ {
		transition, err := kernel.Step(state, reducerNow)
		require.NoError(t, err, "closing preflight must preserve item truth")
		state = transition.State
	}
	scope := onlyFanOutScope(t, state)
	require.Equal(t, WorkItemSucceeded, fanOutItemByKey(scope.FanOut, "a").Status)
	require.Equal(t, WorkItemSucceeded, fanOutItemByKey(scope.FanOut, "b").Status)
	require.Equal(t, ScopeSucceeded, scope.Status)
	require.False(t, scope.FanOut.ResourceFailure)
	require.NoError(t, validateFanOutHardStateBudget("closing preflight", state))
}

func TestFanOutClosingProjectionIncludesPureEndDurableOutputs(t *testing.T) {
	t.Parallel()
	kernel := compilePlanKernel(t, fanOutSwitchChainRuntimePlan(3))
	state := startFanOut(t, kernel, []any{workItemRaw("root", nil, "@dev1")})
	intent, state := stepToMessageIntent(t, kernel, state)
	state = completeIntent(
		t,
		kernel,
		state,
		intent,
		"completed",
		Values{"answer": strings.Repeat("x", 3<<20)},
		nil,
	)
	for index := 0; index < 3; index++ {
		transition, err := kernel.Step(state, reducerNow)
		require.NoError(t, err)
		state = transition.State
		require.NoError(t, validateFanOutHardStateBudget("closing projection step", state))
	}
	scope := onlyFanOutScope(t, state)
	root := fanOutItemByKey(scope.FanOut, "root")
	require.Equal(t, WorkItemSucceeded, root.Status)

	terminal := stepUntilTerminal(t, kernel, state, 4)
	require.Equal(t, StatusSucceeded, terminal.Status)
	require.NoError(t, validateFanOutHardStateBudget("closing projection terminal", terminal))
}

func TestFanOutClosingCausalEffectAndEndNeverCrossStoreBoundary(t *testing.T) {
	t.Parallel()
	kernel := compilePlanKernel(t, fanOutPostCallbackRuntimePlan(2))
	state := startFanOut(t, kernel, []any{workItemRaw("root", nil, "@dev1")})
	itemIntent, state := stepToMessageIntent(t, kernel, state)
	state = completeIntent(t, kernel, state, itemIntent, "completed", Values{
		"answer": strings.Repeat("x", 3<<20),
	}, nil)
	for index := 0; index < 2; index++ {
		transition, err := kernel.Step(state, reducerNow)
		require.NoError(t, err)
		state = transition.State
		require.NoError(t, validateFanOutHardStateBudget("causal item route", state))
	}

	joinStep, err := kernel.Step(state, reducerNow)
	require.NoError(t, err)
	state = joinStep.State
	require.NoError(t, validateFanOutHardStateBudget("causal post ready", state))
	postIntent, state := stepToMessageIntent(t, kernel, state)
	require.True(t, func() bool {
		activation, activationErr := state.activation(postIntent.ActivationID)
		return activationErr == nil && activation.FanOutResourceCausal
	}())
	postCompletion, err := kernel.Complete(state, Completion{
		ActivationID: postIntent.ActivationID,
		Port:         "completed",
		Output:       Values{"ack": "ok"},
	}, reducerNow)
	require.NoError(t, err)
	state = postCompletion.State
	require.NoError(t, validateFanOutHardStateBudget("causal post completion", state))

	terminal, err := kernel.Step(state, reducerNow)
	require.NoError(t, err, "causal End overflow must become a controlled resource outcome")
	require.True(t, terminal.State.Status.Terminal())
	require.NoError(t, validateFanOutHardStateBudget("causal End settlement", terminal.State))
	if terminal.State.Status == StatusFailed {
		require.Equal(t, workItemResourceLimitFailureCode, terminal.State.Failure.Code)
	}
}

func TestFanOutCausalEdgeRejectsOversizedTargetInputAndReplaysCompletion(t *testing.T) {
	t.Parallel()
	kernel := compilePlanKernel(t, fanOutCausalRepeatedBindingRuntimePlan())
	state := startFanOut(t, kernel, []any{workItemRaw("root", nil, "@dev1")})
	itemIntent, state := stepToMessageIntent(t, kernel, state)
	state = completeIntent(t, kernel, state, itemIntent, "completed", Values{
		"answer": "root result",
	}, nil)

	itemRoute, err := kernel.Step(state, reducerNow)
	require.NoError(t, err)
	state = itemRoute.State
	joinStep, err := kernel.Step(state, reducerNow)
	require.NoError(t, err)
	state = joinStep.State
	postIntent, state := stepToMessageIntent(t, kernel, state)
	postActivation, err := state.activation(postIntent.ActivationID)
	require.NoError(t, err)
	require.True(t, postActivation.FanOutResourceCausal)

	completion := Completion{
		ActivationID: postIntent.ActivationID,
		Port:         "completed",
		Output:       Values{"ack": strings.Repeat("x", 2200<<10)},
	}
	settled, err := kernel.Complete(state, completion, reducerNow)
	require.NoError(t, err, "oversized causal target input must take the source failed port")
	require.NoError(t, validateFanOutHardStateBudget("causal oversized target input", settled.State))

	postActivation, err = settled.State.activation(postIntent.ActivationID)
	require.NoError(t, err)
	require.Equal(t, ActivationSucceeded, postActivation.Status)
	require.Equal(t, []string{"failed"}, postActivation.SelectedPorts)
	require.Equal(t, workItemResourceLimitFailureCode, postActivation.Failure.Code)
	require.NotNil(t, postActivation.ResourceRejectedCompletion,
		"the redirected completion retains its compact replay receipt")
	for index := range settled.State.Activations {
		require.NotEqual(t, "wide", settled.State.Activations[index].NodeID,
			"an over-budget target input is never appended to durable State")
	}

	replayed, err := kernel.Complete(settled.State, completion, reducerNow)
	require.NoError(t, err)
	require.Len(t, replayed.Events, 1)
	require.Equal(t, "completion_replayed", replayed.Events[0].Kind)
	require.Len(t, replayed.State.Activations, len(settled.State.Activations))
	replayedPost, err := replayed.State.activation(postIntent.ActivationID)
	require.NoError(t, err)
	require.Equal(t, postActivation.ResourceRejectedCompletion, replayedPost.ResourceRejectedCompletion)
	require.Equal(t, postActivation.SelectedPorts, replayedPost.SelectedPorts)
}

func TestFanOutClosingProjectionBudgetsDurableEffectRequest(t *testing.T) {
	t.Parallel()
	planSource := strings.Replace(
		fanOutPostCallbackRuntimePlan(1),
		"      text: post-process",
		`      text: "{{ input.results }}"`,
		1,
	)
	kernel := compilePlanKernel(t, planSource)
	state := startFanOut(t, kernel, []any{workItemRaw("root", nil, "@dev1")})
	itemIntent, state := stepToMessageIntent(t, kernel, state)
	state = completeIntent(t, kernel, state, itemIntent, "completed", Values{
		"answer": strings.Repeat("x", 2200<<10),
	}, nil)

	itemRoute, err := kernel.Step(state, reducerNow)
	require.NoError(t, err, "closing projection must convert an oversized future effect to an item-local resource failure")
	state = itemRoute.State
	require.NoError(t, validateFanOutHardStateBudget("durable effect projection item route", state))
	scope := onlyFanOutScope(t, state)
	item := fanOutItemByKey(scope.FanOut, "root")
	require.Equal(t, WorkItemFailed, item.Status)
	require.Equal(t, workItemResourceLimitFailureCode, item.Failure.Code)

	joinStep, err := kernel.Step(state, reducerNow)
	require.NoError(t, err)
	state = joinStep.State
	postIntent, state := stepToMessageIntent(t, kernel, state)
	require.NoError(t, validateFanOutIntentBudget("compacted post effect", postIntent))
	require.NoError(t, validateFanOutHardStateBudget("compacted post effect state", state))
}

func TestFanOutRepeatedDependencyBindingFailsOnlyConsumer(t *testing.T) {
	t.Parallel()
	planSource := strings.Replace(
		fanOutRuntimePlan,
		"      dependencies: {type: object, required: true}\n    sendMessage:",
		"      dependencies: {type: object, required: true}\n      dependencies_copy: {type: object, required: true}\n    sendMessage:",
		1,
	)
	planSource = strings.Replace(
		planSource,
		"      dependencies: {ref: item.dependencies}\n  - id: implemented",
		"      dependencies: {ref: item.dependencies}\n      dependencies_copy: {ref: item.dependencies}\n  - id: implemented",
		1,
	)
	kernel := compilePlanKernel(t, planSource)
	state := startFanOut(t, kernel, []any{
		workItemRaw("root", nil, "@dev1"),
		workItemRaw("child", []string{"root"}, "@dev2"),
		workItemRaw("grandchild", []string{"child"}, "@dev1"),
		workItemRaw("independent", nil, "@dev2"),
	})
	rootIntent, state := stepToMessageIntent(t, kernel, state)
	independentIntent, state := stepToMessageIntent(t, kernel, state)
	require.Equal(t, "root", activationItemKey(t, state, rootIntent.ActivationID))
	require.Equal(t, "independent", activationItemKey(t, state, independentIntent.ActivationID))

	state = completeIntent(
		t,
		kernel,
		state,
		rootIntent,
		"completed",
		Values{"answer": strings.Repeat("r", 2<<20)},
		nil,
	)
	scope := onlyFanOutScope(t, state)
	require.Equal(t, WorkItemSucceeded, fanOutItemByKey(scope.FanOut, "root").Status)
	child := fanOutItemByKey(scope.FanOut, "child")
	require.Equal(t, WorkItemFailed, child.Status)
	require.Equal(t, workItemResourceLimitFailureCode, child.Failure.Code)
	require.Empty(t, child.RootActivationID, "the rejected child never gets an activation/effect")
	require.Equal(t, WorkItemBlocked, fanOutItemByKey(scope.FanOut, "grandchild").Status)
	require.Equal(t, WorkItemRunning, fanOutItemByKey(scope.FanOut, "independent").Status)

	state = completeIntent(
		t,
		kernel,
		state,
		independentIntent,
		"completed",
		Values{"answer": "independent result"},
		nil,
	)
	scope = onlyFanOutScope(t, state)
	require.Equal(t, ScopeSucceeded, scope.Status)
	require.Equal(t, WorkItemSucceeded, fanOutItemByKey(scope.FanOut, "independent").Status)
	encoded, err := json.Marshal(state)
	require.NoError(t, err)
	require.LessOrEqual(t, len(encoded), runlimit.MaxRunStateBytes)
}

func TestFanOutRecoveredPendingItemDerivesDependencyView(t *testing.T) {
	t.Parallel()
	kernel := compilePlanKernel(t, strings.Replace(
		fanOutRuntimePlan,
		"maxConcurrency: 2",
		"maxConcurrency: 1",
		1,
	))
	state := startFanOut(t, kernel, []any{
		workItemRaw("root", nil, "@dev1"),
		workItemRaw("occupies_slot", nil, "@dev2"),
		workItemRaw("child", []string{"root"}, "@dev1"),
	})
	rootIntent, state := stepToMessageIntent(t, kernel, state)
	state = completeIntent(t, kernel, state, rootIntent, "completed", Values{"answer": "root result"}, nil)
	scope := onlyFanOutScope(t, state)
	require.Equal(t, WorkItemRunning, fanOutItemByKey(scope.FanOut, "occupies_slot").Status)
	require.Equal(t, WorkItemPending, fanOutItemByKey(scope.FanOut, "child").Status)

	restored, err := JSONRoundTrip(state)
	require.NoError(t, err)
	occupierIntent, restored := stepToMessageIntent(t, kernel, restored)
	restored = completeIntent(t, kernel, restored, occupierIntent, "completed", Values{"answer": "slot done"}, nil)
	childActivation := activationForItem(t, restored, "child", ActivationReady)
	dependencies := childActivation.Input[plan.FanOutItemFieldDependencies].(map[string]any)
	require.Equal(
		t,
		map[string]any{"answer": "root result"},
		dependencies["root"].(map[string]any)["value"],
		"recovery reconstructs the launch view from the predecessor's single durable Value",
	)
}