package orchestrator

import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"strings"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/core/domain"
	runstore "github.com/tidbcloud/AgentTeam/internal/store/workflow/run"
	"github.com/tidbcloud/AgentTeam/internal/workflow/plan"
	runkernel "github.com/tidbcloud/AgentTeam/internal/workflow/run"
)

func TestEngineStateFailureFallsBackToBoundedFatalProjection(t *testing.T) {
	now := time.Date(2026, 7, 21, 5, 55, 0, 0, time.UTC)
	material := callbackMaterial(t)
	graph, err := runkernel.DecodeArtifact(material.ArtifactJSON)
	require.NoError(t, err)
	run := claimedRuntimeRun(now, []byte(`{}`), 2)
	state := &runkernel.State{
		Status: runkernel.StatusRunning, Input: runkernel.Values{},
		Variables:   runkernel.Values{"nearLimit": strings.Repeat("x", runstore.MaxRunStateBytes)},
		Activations: []runkernel.Activation{}, Scopes: []runkernel.Scope{},
	}
	repository := &engineRepositoryStub{}
	engine, err := NewEngine(repository, materialLoaderStub{material: material}, EngineOptions{WorkerID: "worker"})
	require.NoError(t, err)
	claim := runstore.ClaimedRun{
		Run: run, Fence: runstore.TurnFence{WorkspaceID: run.WorkspaceID, RunID: run.ID, LeaseToken: run.LeaseToken},
	}
	require.NoError(t, engine.saveStateFailure(
		context.Background(), claim, graph, state, nil, nil, "manual", "run.reducer_failed", fmt.Errorf("deterministic reducer failure"), now,
	))
	require.Len(t, repository.failed, 1)
	require.Equal(t, "run.runtime_state_limit_exceeded", repository.failed[0].ErrorCode)
	require.LessOrEqual(t, len(repository.failed[0].StateJSON), runstore.MaxRunStateBytes)
}

func TestEngineStateFailureConvertsOversizedOutputToDurableFatalProjection(t *testing.T) {
	now := time.Date(2026, 7, 21, 5, 56, 0, 0, time.UTC)
	material := callbackMaterial(t)
	graph, err := runkernel.DecodeArtifact(material.ArtifactJSON)
	require.NoError(t, err)
	run := claimedRuntimeRun(now, []byte(`{}`), 2)
	state := &runkernel.State{
		Status: runkernel.StatusRunning, Input: runkernel.Values{}, Variables: runkernel.Values{},
		Output:      runkernel.Values{"oversized": strings.Repeat("x", runstore.MaxRunPayloadBytes)},
		Activations: []runkernel.Activation{}, Scopes: []runkernel.Scope{},
	}
	repository := &engineRepositoryStub{}
	engine, err := NewEngine(repository, materialLoaderStub{material: material}, EngineOptions{WorkerID: "worker"})
	require.NoError(t, err)
	claim := runstore.ClaimedRun{
		Run: run, Fence: runstore.TurnFence{WorkspaceID: run.WorkspaceID, RunID: run.ID, LeaseToken: run.LeaseToken},
	}
	require.NoError(t, engine.saveStateFailure(
		context.Background(), claim, graph, state, nil, nil, "manual", "run.reducer_failed", fmt.Errorf("deterministic reducer failure"), now,
	))
	require.Len(t, repository.failed, 1)
	require.Equal(t, "run.runtime_payload_limit_exceeded", repository.failed[0].ErrorCode)
}

func TestEngineStateFailureConvertsStructurallyOversizedOutputToDurableFatalProjection(t *testing.T) {
	now := time.Date(2026, 7, 21, 5, 57, 0, 0, time.UTC)
	material := callbackMaterial(t)
	graph, err := runkernel.DecodeArtifact(material.ArtifactJSON)
	require.NoError(t, err)
	run := claimedRuntimeRun(now, []byte(`{}`), 2)
	wideValues := make([]any, 9_000)
	for index := range wideValues {
		wideValues[index] = index
	}
	state := &runkernel.State{
		Status: runkernel.StatusRunning, Input: runkernel.Values{}, Variables: runkernel.Values{},
		Output:      runkernel.Values{"wide": wideValues},
		Activations: []runkernel.Activation{}, Scopes: []runkernel.Scope{},
	}
	encodedOutput, err := json.Marshal(state.Output)
	require.NoError(t, err)
	require.Less(t, len(encodedOutput), runstore.MaxRunPayloadBytes, "the structural budget, not the byte budget, must reject this payload")
	repository := &engineRepositoryStub{}
	engine, err := NewEngine(repository, materialLoaderStub{material: material}, EngineOptions{WorkerID: "worker"})
	require.NoError(t, err)
	claim := runstore.ClaimedRun{
		Run: run, Fence: runstore.TurnFence{WorkspaceID: run.WorkspaceID, RunID: run.ID, LeaseToken: run.LeaseToken},
	}
	require.NoError(t, engine.saveStateFailure(
		context.Background(), claim, graph, state, nil, nil, "manual", "run.reducer_failed", fmt.Errorf("deterministic reducer failure"), now,
	))
	require.Empty(t, repository.saved, "a Store BadRequest must not poison-retry the claimed turn")
	require.Len(t, repository.failed, 1)
	require.Equal(t, "run.runtime_payload_limit_exceeded", repository.failed[0].ErrorCode)
}

func TestUnrecoverableFailureSnapshotFallsBackWhenFailureMetadataCrossesStateStructureBudget(t *testing.T) {
	// The initialized snapshot fits exactly within the aggregate value budget.
	// Adding Run and live-Activation Failure objects crosses it, so the fatal
	// projection must use its minimal snapshot rather than handing a payload to
	// FailClaimedRun that the Store will reject forever.
	padding := make([]int, runstore.MaxRunStateJSONValues-20)
	state := &runkernel.State{
		Status: runkernel.StatusRunning,
		Input:  runkernel.Values{},
		Variables: runkernel.Values{
			"padding": padding,
		},
		Activations: []runkernel.Activation{{
			ID: "a1", Sequence: 1, NodeID: "end", InputPort: "in", Input: runkernel.Values{},
			SelectedPorts: []string{}, Status: runkernel.ActivationReady,
			Lineage: []runkernel.ScopeRef{}, OwnedScopeIDs: []string{},
		}},
		Scopes: []runkernel.Scope{}, NextActivation: 1,
	}
	encoded, err := json.Marshal(state)
	require.NoError(t, err)
	encoded, err = runstore.DeterministicStateJSON(encoded)
	require.NoError(t, err, "fixture must fit before failure metadata is added")
	run := runstore.Run{InputJSON: json.RawMessage(`{}`), StateJSON: encoded}

	failedJSON, _, err := unrecoverableFailureSnapshot(run, "run.reducer_failed", fmt.Errorf("deterministic failure"))
	require.NoError(t, err)
	_, err = runstore.DeterministicStateJSON(failedJSON)
	require.NoError(t, err, "fatal snapshot itself must pass the exact Store boundary")
	failed, initialized, err := decodeRunState(failedJSON)
	require.NoError(t, err)
	require.True(t, initialized)
	require.Equal(t, runkernel.StatusFailed, failed.Status)
	require.Empty(t, failed.Activations, "structurally over-budget preserved state must fall back to minimal")
	require.Empty(t, failed.Variables)
}

type engineRepositoryStub struct {
	claims        []runstore.ClaimedRun
	claimRequests []runstore.ClaimRunnableParams
	activations   []runstore.Activation
	effects       []runstore.Effect
	saved         []runstore.SaveTurnParams
	failed        []runstore.FailClaimedRunParams
}

func (stub *engineRepositoryStub) ClaimRunnableRuns(_ context.Context, params runstore.ClaimRunnableParams) ([]runstore.ClaimedRun, error) {
	stub.claimRequests = append(stub.claimRequests, params)
	return append([]runstore.ClaimedRun(nil), stub.claims...), nil
}

func (stub *engineRepositoryStub) ListActivations(_ context.Context, params runstore.ListChildrenParams) ([]runstore.Activation, error) {
	return filterActivationsAfter(stub.activations, params.AfterID, params.Limit), nil
}

func (stub *engineRepositoryStub) ListEffects(_ context.Context, params runstore.ListChildrenParams) ([]runstore.Effect, error) {
	return filterEffectsAfter(stub.effects, params.AfterID, params.Limit), nil
}

func (stub *engineRepositoryStub) SaveTurn(_ context.Context, params runstore.SaveTurnParams) (runstore.SaveTurnResult, error) {
	stub.saved = append(stub.saved, params)
	return runstore.SaveTurnResult{}, nil
}

func (stub *engineRepositoryStub) FailClaimedRun(_ context.Context, params runstore.FailClaimedRunParams) (runstore.FailClaimedRunResult, error) {
	stub.failed = append(stub.failed, params)
	return runstore.FailClaimedRunResult{}, nil
}

type materialLoaderStub struct {
	material Material
	err      error
}

func (stub materialLoaderStub) LoadRunMaterial(context.Context, runstore.Run) (Material, error) {
	return stub.material, stub.err
}

func TestEngineRunOnceUsesFreshProcessingTimeForEachClaim(t *testing.T) {
	scanAt := time.Date(2026, 7, 21, 5, 59, 58, 0, time.UTC)
	firstAt := scanAt.Add(time.Second)
	secondAt := scanAt.Add(2 * time.Second)
	times := []time.Time{scanAt, firstAt, secondAt}
	clockCalls := 0
	now := func() time.Time {
		require.Less(t, clockCalls, len(times))
		value := times[clockCalls]
		clockCalls++
		return value
	}
	firstRun := claimedRuntimeRun(scanAt, []byte(`{}`), 2)
	secondRun := claimedRuntimeRun(scanAt, []byte(`{}`), 2)
	secondRun.ID = "wri_second"
	secondRun.LeaseToken = "lease_second"
	repository := &engineRepositoryStub{claims: []runstore.ClaimedRun{
		{
			Run: firstRun,
			Fence: runstore.TurnFence{
				WorkspaceID: firstRun.WorkspaceID, RunID: firstRun.ID, LeaseToken: firstRun.LeaseToken,
			},
		},
		{
			Run: secondRun,
			Fence: runstore.TurnFence{
				WorkspaceID: secondRun.WorkspaceID, RunID: secondRun.ID, LeaseToken: secondRun.LeaseToken,
			},
		},
	}}
	engine, err := NewEngine(repository, materialLoaderStub{material: callbackMaterial(t)}, EngineOptions{WorkerID: "worker", Now: now})
	require.NoError(t, err)

	outcome, err := engine.RunOnce(context.Background())
	require.NoError(t, err)
	require.Equal(t, RunOnceOutcome{Claimed: 2, Succeeded: 2}, outcome)
	require.Equal(t, 3, clockCalls)
	require.Len(t, repository.claimRequests, 1)
	require.Equal(t, scanAt, repository.claimRequests[0].Now)
	require.Len(t, repository.saved, 2)
	require.Equal(t, firstAt, repository.saved[0].Now)
	require.Equal(t, secondAt, repository.saved[1].Now)
}

func TestEngineInitialClaimReducesToDurableCallbackWait(t *testing.T) {
	now := time.Date(2026, 7, 21, 6, 0, 0, 0, time.UTC)
	material := callbackMaterial(t)
	run := claimedRuntimeRun(now, []byte(`{}`), 2)
	repository := &engineRepositoryStub{}
	engine, err := NewEngine(repository, materialLoaderStub{material: material}, EngineOptions{WorkerID: "worker", Now: func() time.Time { return now }})
	require.NoError(t, err)
	require.NoError(t, engine.ProcessClaim(context.Background(), runstore.ClaimedRun{
		Run: run, Fence: runstore.TurnFence{WorkspaceID: run.WorkspaceID, RunID: run.ID, LeaseToken: run.LeaseToken},
	}, now))
	require.Len(t, repository.saved, 1)
	saved := repository.saved[0]
	require.Equal(t, runstore.RunStatusWaiting, saved.Status)
	require.Len(t, saved.Activations, 1)
	require.Len(t, saved.Effects, 1)
	require.Contains(t, eventTypes(saved.Events), initialRunStartedEvent)
	require.Contains(t, eventTypes(saved.Events), "effect_requested")
	request, err := DecodeEffectRequest(saved.Effects[0].Value.RequestJSON)
	require.NoError(t, err)
	require.Equal(t, "@agent answer in 9 characters or more", request.Message.Content)
}

func TestEngineConsumesCompletedCallbackAndFinishesPureTail(t *testing.T) {
	now := time.Date(2026, 7, 21, 6, 0, 0, 0, time.UTC)
	material := callbackMaterial(t)
	graph, err := runkernel.DecodeArtifact(material.ArtifactJSON)
	require.NoError(t, err)
	kernel, err := runkernel.NewKernel(graph)
	require.NoError(t, err)
	state, err := kernel.Start(runkernel.Values{"channel": "#workflow"})
	require.NoError(t, err)
	waiting, err := kernel.Step(state, now)
	require.NoError(t, err)
	stateJSON, err := json.Marshal(waiting.State)
	require.NoError(t, err)
	run := claimedRuntimeRun(now.Add(time.Second), stateJSON, 3)
	seed, err := ProjectTurn(ProjectTurnInput{
		Run: run, Fence: runstore.TurnFence{WorkspaceID: run.WorkspaceID, RunID: run.ID, LeaseToken: run.LeaseToken},
		Graph: graph, Transition: waiting, SourceKind: "manual", Now: now,
	})
	require.NoError(t, err)
	activation := seed.Activations[0].Value
	activation.CreatedAt = now
	effect := seed.Effects[0].Value
	completedAt := now.Add(time.Second)
	effect.Status = runstore.EffectStatusCompleted
	effect.CompletedAt = &completedAt
	effect.ResultJSON, err = EncodeEffectResult(EffectResult{
		Port: "completed", Output: runkernel.Values{"reply": "done"}, MessageID: "msg", WakeAgentIDs: []string{"agt"}, CompletedAt: &completedAt,
	})
	require.NoError(t, err)
	repository := &engineRepositoryStub{activations: []runstore.Activation{activation}, effects: []runstore.Effect{effect}}
	engine, err := NewEngine(repository, materialLoaderStub{material: material}, EngineOptions{WorkerID: "worker"})
	require.NoError(t, err)
	require.NoError(t, engine.ProcessClaim(context.Background(), runstore.ClaimedRun{
		Run: run, Fence: runstore.TurnFence{WorkspaceID: run.WorkspaceID, RunID: run.ID, LeaseToken: run.LeaseToken},
	}, completedAt))
	require.Len(t, repository.saved, 1)
	saved := repository.saved[0]
	require.Equal(t, runstore.RunStatusSucceeded, saved.Status)
	require.JSONEq(t, `{"reply":"done"}`, string(saved.OutputJSON))
	require.Len(t, saved.Effects, 0)
	require.Contains(t, eventTypes(saved.Events), "activation_completed")
	require.Contains(t, eventTypes(saved.Events), "run_succeeded")
}

func TestEngineDoesNotTimeOutOverduePendingCallback(t *testing.T) {
	now := time.Date(2026, 7, 21, 6, 0, 0, 0, time.UTC)
	material := callbackMaterial(t)
	graph, err := runkernel.DecodeArtifact(material.ArtifactJSON)
	require.NoError(t, err)
	kernel, err := runkernel.NewKernel(graph)
	require.NoError(t, err)
	state, err := kernel.Start(runkernel.Values{"channel": "#workflow"})
	require.NoError(t, err)
	waiting, err := kernel.Step(state, now)
	require.NoError(t, err)
	stateJSON, err := json.Marshal(waiting.State)
	require.NoError(t, err)
	run := claimedRuntimeRun(now.Add(31*time.Minute), stateJSON, 3)
	seed, err := ProjectTurn(ProjectTurnInput{
		Run: run, Fence: runstore.TurnFence{WorkspaceID: run.WorkspaceID, RunID: run.ID, LeaseToken: run.LeaseToken},
		Graph: graph, Transition: waiting, SourceKind: "manual", Now: now,
	})
	require.NoError(t, err)
	activation := seed.Activations[0].Value
	activation.CreatedAt = now
	effect := seed.Effects[0].Value
	require.Equal(t, runstore.EffectStatusPending, effect.Status)
	repository := &engineRepositoryStub{activations: []runstore.Activation{activation}, effects: []runstore.Effect{effect}}
	engine, err := NewEngine(repository, materialLoaderStub{material: material}, EngineOptions{WorkerID: "worker"})
	require.NoError(t, err)
	require.NoError(t, engine.ProcessClaim(context.Background(), runstore.ClaimedRun{
		Run: run, Fence: runstore.TurnFence{WorkspaceID: run.WorkspaceID, RunID: run.ID, LeaseToken: run.LeaseToken},
	}, now.Add(31*time.Minute)))
	saved := repository.saved[0]
	require.Equal(t, runstore.RunStatusWaiting, saved.Status,
		"an authored callback deadline must not fail the Run")
	require.Empty(t, saved.Effects,
		"an overdue callback must not be mutated into a timeout completion")
	for _, event := range saved.Events {
		require.NotEqual(t, "effect_timed_out", event.Type)
	}
	stateAfter := decodeSavedRunState(t, saved.StateJSON)
	require.Equal(t, runkernel.ActivationWaiting, stateAfter.Activations[0].Status,
		"the callback activation must remain waiting for a receipt or infrastructure failure")
}

func TestAvailableEffectCompletionDoesNotTimeOutLiveCallback(t *testing.T) {
	t.Parallel()

	createdAt := time.Date(2026, 7, 31, 9, 0, 0, 0, time.UTC)
	effect := claimedMessageEffect(t, createdAt, true)
	effect.Status = runstore.EffectStatusAwaiting
	deliveryScope := domain.MessageScopeValue{
		Kind: domain.MessageScopeKindTask, ScopeID: "tsk_callback_work",
	}
	var err error
	effect.ResultJSON, err = EncodeEffectResult(EffectResult{
		MessageID: "msg_callback",
		WakeAgentIDs: []string{
			"agt_watcher", "agt_callback", "agt_watcher",
		},
		CallbackResponderAgentID: "agt_callback",
		DeliveryScope:            &deliveryScope,
	})
	require.NoError(t, err)

	state := &runkernel.State{Activations: []runkernel.Activation{{
		ID: "a1", Status: runkernel.ActivationWaiting,
	}}}
	for _, status := range []string{
		runstore.EffectStatusPending,
		runstore.EffectStatusDelivered,
		runstore.EffectStatusAwaiting,
	} {
		candidate := effect
		candidate.Status = status
		completion, available, err := availableEffectCompletion(state, candidate)
		require.NoError(t, err)
		require.False(t, available,
			"a live %s callback must never reduce to a timeout completion", status)
		require.Empty(t, completion.Port)
	}
}

func TestAvailableEffectCompletionReducesOnlyTerminalEffects(t *testing.T) {
	t.Parallel()

	now := time.Date(2026, 7, 31, 9, 30, 0, 0, time.UTC)
	state := &runkernel.State{Activations: []runkernel.Activation{{
		ID: "a1", Status: runkernel.ActivationWaiting,
	}}}

	t.Run("completed receipt", func(t *testing.T) {
		effect := claimedMessageEffect(t, now, true)
		effect.Status = runstore.EffectStatusCompleted
		completedAt := now.Add(time.Second)
		effect.CompletedAt = &completedAt
		encoded, err := EncodeEffectResult(EffectResult{
			Port: "completed", Output: runkernel.Values{"reply": "done"},
			MessageID: "msg_completed", WakeAgentIDs: []string{"agt_callback"},
			CallbackResponderAgentID: "agt_callback", CompletedAt: &completedAt,
		})
		require.NoError(t, err)
		effect.ResultJSON = encoded
		completion, available, err := availableEffectCompletion(state, effect)
		require.NoError(t, err)
		require.True(t, available)
		require.Equal(t, "completed", completion.Port)
		require.Equal(t, "agt_callback", completion.CallbackResponderAgentID)
	})

	t.Run("infrastructure failure", func(t *testing.T) {
		effect := claimedMessageEffect(t, now, true)
		effect.Status = runstore.EffectStatusFailed
		effect.ErrorCode = "message.delivery_failed"
		effect.ResultJSON = nil
		completion, available, err := availableEffectCompletion(state, effect)
		require.NoError(t, err)
		require.True(t, available)
		require.Equal(t, "failed", completion.Port)
		require.Equal(t, "message.delivery_failed", completion.Failure.Code)
	})
}

func TestEngineKeepsAwaitingCallbackWaitingPastAuthoredDeadline(t *testing.T) {
	t.Parallel()

	dispatchedAt := time.Date(2026, 7, 31, 9, 0, 0, 0, time.UTC)
	effect := claimedMessageEffect(t, dispatchedAt, true)
	effect.Status = runstore.EffectStatusAwaiting
	deliveryScope := domain.MessageScopeValue{
		Kind: domain.MessageScopeKindTask, ScopeID: "tsk_callback_work",
	}
	var err error
	effect.ResultJSON, err = EncodeEffectResult(EffectResult{
		MessageID: "msg_callback",
		WakeAgentIDs: []string{
			"agt_watcher", "agt_callback", "agt_watcher",
		},
		CallbackResponderAgentID: "agt_callback",
		DeliveryScope:            &deliveryScope,
	})
	require.NoError(t, err)

	state := &runkernel.State{Activations: []runkernel.Activation{{
		ID: "a1", Status: runkernel.ActivationWaiting,
	}}}
	completion, available, err := availableEffectCompletion(state, effect)
	require.NoError(t, err)
	require.False(t, available,
		"an awaiting callback past its authored deadline must stay waiting for the Agent reply")
	require.Empty(t, completion.Port)
	require.Empty(t, completion.CallbackResponderAgentID)
}

func TestEngineWaitsForInfrastructureFailureOnPendingCallback(t *testing.T) {
	t.Parallel()

	createdAt := time.Date(2026, 7, 31, 10, 0, 0, 0, time.UTC)
	effect := claimedMessageEffect(t, createdAt, true)
	effect.Status = runstore.EffectStatusPending
	effect.ResultJSON = nil

	state := &runkernel.State{Activations: []runkernel.Activation{{
		ID: "a1", Status: runkernel.ActivationWaiting,
	}}}
	completion, available, err := availableEffectCompletion(state, effect)
	require.NoError(t, err)
	require.False(t, available,
		"a pending callback with no committed receipt must wait for dispatch or an infrastructure failure")
	require.Empty(t, completion.Port)

	// A definitive infrastructure delivery failure still reduces through the
	// failed port; only authored callback deadlines were removed.
	failed := effect
	failed.Status = runstore.EffectStatusFailed
	failed.ErrorCode = "message.delivery_failed"
	failed.ResultJSON = nil
	completion, available, err = availableEffectCompletion(state, failed)
	require.NoError(t, err)
	require.True(t, available)
	require.Equal(t, "failed", completion.Port)
	require.Equal(t, "message.delivery_failed", completion.Failure.Code)
	require.Empty(t, completion.CallbackResponderAgentID)
}

func TestEnginePersistsOnlyFirstReadyFixedTaskCallbackBeforeDeliveryReceipt(t *testing.T) {
	now := time.Date(2026, 7, 31, 11, 0, 0, 0, time.UTC)
	_, _, saved := runFirstTaskWatchBarrierTurn(t, now, "all")

	require.Equal(t, runstore.RunStatusRunning, saved.Status)
	inserts := insertedTaskWatchEffects(t, saved.Effects)
	require.Len(t, inserts, 1,
		"one reducer turn must not persist two ready fixed-Task callbacks before either delivery receipt exists")
	request, err := DecodeEffectRequest(inserts[0].RequestJSON)
	require.NoError(t, err)
	require.Equal(t, "a", request.Intent.NodeID)
	require.NotNil(t, request.Intent.TaskWatchHandoff)
	require.Nil(t, request.Intent.Deadline, "send_message carries no runtime deadline")
	require.Equal(t, now.Add(parkedRunDelay).UTC(), saved.RunnableAt.UTC(),
		"the ready sibling must not busy-loop the Run while the first Task callback is awaiting dispatch")
	require.True(t, saved.RunnableAt.After(now))

	state := decodeSavedRunState(t, saved.StateJSON)
	a := requireRunActivationByNode(t, state, "a")
	b := requireRunActivationByNode(t, state, "b")
	require.Equal(t, runkernel.ActivationWaiting, a.Status)
	require.NotNil(t, a.CallbackTaskWatch)
	require.Empty(t, a.CallbackTaskWatch.ResponderAgentID)
	require.Equal(t, runkernel.ActivationReady, b.Status,
		"the sibling remains ready but must not emit its effect until the delivery receipt is observed")
}

func TestEngineObservesTaskWatchReceiptBeforeDispatchingReadySibling(t *testing.T) {
	now := time.Date(2026, 7, 31, 11, 30, 0, 0, time.UTC)
	material, firstRun, firstSaved := runFirstTaskWatchBarrierTurn(t, now, "all")
	persistedEffects := insertedEffectMutations(firstSaved.Effects)
	firstEffects := insertedTaskWatchEffects(t, firstSaved.Effects)
	require.Len(t, firstEffects, 1)
	first := firstEffects[0]
	first.Status = runstore.EffectStatusAwaiting
	deliveredAt := now.Add(time.Second)
	first.DeliveredAt = &deliveredAt
	taskScope := taskWatchBarrierScope()
	var err error
	first.ResultJSON, err = EncodeEffectResult(EffectResult{
		MessageID: "msg_a", WakeAgentIDs: []string{"agt_a", "agt_watcher"},
		CallbackResponderAgentID: "agt_a", DeliveryScope: &taskScope,
	})
	require.NoError(t, err)
	replaceEffectByID(t, persistedEffects, first)

	secondRun := resumedTaskWatchBarrierRun(firstRun, firstSaved, deliveredAt)
	repository := &engineRepositoryStub{
		activations: activationMutationValues(firstSaved.Activations),
		effects:     persistedEffects,
	}
	engine, err := NewEngine(
		repository,
		materialLoaderStub{material: material},
		EngineOptions{WorkerID: "worker"},
	)
	require.NoError(t, err)
	require.NoError(t, engine.ProcessClaim(context.Background(), claimedRun(secondRun), deliveredAt))
	require.Len(t, repository.saved, 1)
	saved := repository.saved[0]

	state := decodeSavedRunState(t, saved.StateJSON)
	a := requireRunActivationByNode(t, state, "a")
	b := requireRunActivationByNode(t, state, "b")
	require.Equal(t, runkernel.ActivationWaiting, a.Status)
	require.Equal(t, "agt_a", a.CallbackTaskWatch.ResponderAgentID,
		"the committed receipt must freeze the first branch before its sibling is projected")
	require.Equal(t, runkernel.ActivationWaiting, b.Status)
	require.NotNil(t, b.CallbackTaskWatch)
	require.Empty(t, b.CallbackTaskWatch.ResponderAgentID)

	secondEffects := insertedTaskWatchEffects(t, saved.Effects)
	require.Len(t, secondEffects, 1)
	secondRequest, err := DecodeEffectRequest(secondEffects[0].RequestJSON)
	require.NoError(t, err)
	require.Equal(t, "b", secondRequest.Intent.NodeID)
	require.NotNil(t, secondRequest.Intent.TaskWatchHandoff)
	require.Equal(t, taskScope, secondRequest.Intent.TaskWatchHandoff.TaskScope)
	require.Empty(t, secondRequest.Intent.TaskWatchHandoff.CausalPredecessorAgentIDs,
		"parallel siblings do not hand off or unwatch one another")
	require.Nil(t, first.DeadlineAt, "send_message carries no runtime deadline")
	require.Equal(t, deliveredAt.Add(parkedRunDelay).UTC(), saved.RunnableAt.UTC(),
		"the Run remains parked at the private callback parking horizon, not at the ready sibling")
	require.True(t, saved.RunnableAt.After(deliveredAt))

	events := eventTypes(saved.Events)
	observedIndex := indexOfString(events, "callback_responder_observed")
	requestedIndex := indexOfString(events, "effect_requested")
	require.NotEqual(t, -1, observedIndex)
	require.NotEqual(t, -1, requestedIndex)
	require.Less(t, observedIndex, requestedIndex,
		"receipt observation must be reduced before the sibling callback starts")
}

func TestEngineRestoresUndeliveredTaskWatchProjectionBeforeReadySibling(t *testing.T) {
	for _, test := range []struct {
		name      string
		configure func(*runstore.Effect, time.Time) time.Time
		wantPort  string
	}{
		{
			name: "failed before delivery",
			configure: func(effect *runstore.Effect, now time.Time) time.Time {
				effect.Status = runstore.EffectStatusFailed
				effect.ErrorCode = "message.delivery_failed"
				effect.ResultJSON = nil
				return now.Add(time.Second)
			},
			wantPort: "failed",
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			now := time.Date(2026, 7, 31, 12, 0, 0, 0, time.UTC)
			material, firstRun, firstSaved := runFirstTaskWatchBarrierTurn(t, now, "all")
			persistedEffects := insertedEffectMutations(firstSaved.Effects)
			firstEffects := insertedTaskWatchEffects(t, firstSaved.Effects)
			require.Len(t, firstEffects, 1)
			first := firstEffects[0]
			processAt := test.configure(&first, now)
			replaceEffectByID(t, persistedEffects, first)

			secondRun := resumedTaskWatchBarrierRun(firstRun, firstSaved, processAt)
			repository := &engineRepositoryStub{
				activations: activationMutationValues(firstSaved.Activations),
				effects:     persistedEffects,
			}
			engine, err := NewEngine(
				repository,
				materialLoaderStub{material: material},
				EngineOptions{WorkerID: "worker"},
			)
			require.NoError(t, err)
			require.NoError(t, engine.ProcessClaim(context.Background(), claimedRun(secondRun), processAt))
			require.Len(t, repository.saved, 1)
			saved := repository.saved[0]

			state := decodeSavedRunState(t, saved.StateJSON)
			a := requireRunActivationByNode(t, state, "a")
			b := requireRunActivationByNode(t, state, "b")
			require.Equal(t, runkernel.ActivationSucceeded, a.Status)
			require.Equal(t, []string{test.wantPort}, a.SelectedPorts)
			require.NotNil(t, a.CallbackTaskWatch)
			require.Empty(t, a.CallbackTaskWatch.ResponderAgentID)
			require.Empty(t, a.TaskWatchFrontier,
				"a pre-delivery terminal outcome must restore the provisional frontier")
			require.Equal(t, runkernel.ActivationWaiting, b.Status)

			secondEffects := insertedTaskWatchEffects(t, saved.Effects)
			require.Len(t, secondEffects, 1)
			secondRequest, err := DecodeEffectRequest(secondEffects[0].RequestJSON)
			require.NoError(t, err)
			require.Equal(t, "b", secondRequest.Intent.NodeID)
			require.NotNil(t, secondRequest.Intent.TaskWatchHandoff)
			require.Empty(t, secondRequest.Intent.TaskWatchHandoff.CausalPredecessorAgentIDs,
				"the undelivered first branch cannot become a handoff predecessor")
		})
	}
}

func TestEngineObservesAwaitingReceiptForCanceledJoinAnyLoser(t *testing.T) {
	now := time.Date(2026, 7, 31, 13, 0, 0, 0, time.UTC)
	material := taskWatchBarrierMaterial(t, "any")
	graph, err := runkernel.DecodeArtifact(material.ArtifactJSON)
	require.NoError(t, err)
	kernel, err := runkernel.NewKernel(graph)
	require.NoError(t, err)
	state, err := kernel.Start(taskWatchBarrierInputValues())
	require.NoError(t, err)

	winner, state := stepToRuntimeTaskWatchIntent(t, kernel, state, now)
	observedWinner, err := kernel.ObserveCallbackResponder(state, winner.ActivationID, "agt_a")
	require.NoError(t, err)
	state = observedWinner.State
	loser, state := stepToRuntimeTaskWatchIntent(t, kernel, state, now)
	require.Equal(t, "a", winner.NodeID)
	require.Equal(t, "b", loser.NodeID)
	loserActivation := requireRunActivationByNode(t, state, "b")
	require.Equal(t, runkernel.ActivationWaiting, loserActivation.Status)
	require.NotNil(t, loserActivation.CallbackTaskWatch)
	require.Empty(t, loserActivation.CallbackTaskWatch.ResponderAgentID)

	run := claimedRuntimeRun(now.Add(time.Second), nil, 3)
	run.InputJSON = taskWatchBarrierInputJSON()
	run.Status = runstore.RunStatusWaiting
	stateJSON, err := json.Marshal(state)
	require.NoError(t, err)
	run.StateJSON = stateJSON

	seed, err := ProjectTurn(ProjectTurnInput{
		Run: run,
		Fence: runstore.TurnFence{
			WorkspaceID: run.WorkspaceID, RunID: run.ID, LeaseToken: run.LeaseToken,
		},
		Graph: graph,
		Transition: runkernel.Transition{
			State: state, Intents: []runkernel.Intent{winner, loser}, Events: []runkernel.Event{},
		},
		SourceKind: "manual", Now: now,
	})
	require.NoError(t, err)
	callbackEffects := insertedTaskWatchEffects(t, seed.Effects)
	require.Len(t, callbackEffects, 2)
	var winnerEffect, loserEffect runstore.Effect
	for _, effect := range callbackEffects {
		request, decodeErr := DecodeEffectRequest(effect.RequestJSON)
		require.NoError(t, decodeErr)
		switch request.Intent.NodeID {
		case "a":
			winnerEffect = effect
		case "b":
			loserEffect = effect
		}
	}
	require.NotEmpty(t, winnerEffect.ID)
	require.NotEmpty(t, loserEffect.ID)
	completedAt := now.Add(time.Second)
	winnerEffect.Status = runstore.EffectStatusCompleted
	winnerEffect.CompletedAt = &completedAt
	taskScope := taskWatchBarrierScope()
	winnerEffect.ResultJSON, err = EncodeEffectResult(EffectResult{
		Port: "completed", Output: taskWatchBarrierCallbackOutput("winner"),
		MessageID: "msg_a", WakeAgentIDs: []string{"agt_a"},
		CallbackResponderAgentID: "agt_a", DeliveryScope: &taskScope,
		CompletedAt: &completedAt,
	})
	require.NoError(t, err)
	loserEffect.Status = runstore.EffectStatusAwaiting
	deliveredAt := completedAt
	loserEffect.DeliveredAt = &deliveredAt
	loserEffect.ResultJSON, err = EncodeEffectResult(EffectResult{
		MessageID: "msg_b", WakeAgentIDs: []string{"agt_b"},
		CallbackResponderAgentID: "agt_b", DeliveryScope: &taskScope,
	})
	require.NoError(t, err)

	repository := &engineRepositoryStub{
		activations: activationMutationValues(seed.Activations),
		effects:     []runstore.Effect{winnerEffect, loserEffect},
	}
	engine, err := NewEngine(
		repository,
		materialLoaderStub{material: material},
		EngineOptions{WorkerID: "worker"},
	)
	require.NoError(t, err)
	require.NoError(t, engine.ProcessClaim(context.Background(), claimedRun(run), deliveredAt))
	require.Len(t, repository.saved, 1)
	saved := repository.saved[0]
	observedState := decodeSavedRunState(t, saved.StateJSON)
	observedLoser := requireRunActivationByNode(t, observedState, "b")
	require.Equal(t, runkernel.ActivationCanceled, observedLoser.Status)
	require.Equal(t, "agt_b", observedLoser.CallbackTaskWatch.ResponderAgentID,
		"the awaiting loser receipt must be observed before the winner completion cancels its branch")
	require.Contains(t, eventTypes(saved.Events), "callback_responder_observed")
	canceledEffects := effectMutationsWithStatus(saved.Effects, runstore.EffectStatusCanceled)
	require.Len(t, canceledEffects, 1)
	require.Equal(t, runstore.EffectStatusAwaiting, canceledEffects[0].ExpectedStatus)
}

func TestEngineDoesNotTimeoutEffectWithCommittedDispatchLease(t *testing.T) {
	now := time.Date(2026, 7, 21, 6, 15, 0, 0, time.UTC)
	material := callbackMaterial(t)
	graph, err := runkernel.DecodeArtifact(material.ArtifactJSON)
	require.NoError(t, err)
	kernel, err := runkernel.NewKernel(graph)
	require.NoError(t, err)
	state, err := kernel.Start(runkernel.Values{"channel": "#workflow"})
	require.NoError(t, err)
	waiting, err := kernel.Step(state, now)
	require.NoError(t, err)
	stateJSON, err := json.Marshal(waiting.State)
	require.NoError(t, err)
	run := claimedRuntimeRun(now.Add(31*time.Minute), stateJSON, 3)
	seed, err := ProjectTurn(ProjectTurnInput{
		Run: run, Fence: runstore.TurnFence{WorkspaceID: run.WorkspaceID, RunID: run.ID, LeaseToken: run.LeaseToken},
		Graph: graph, Transition: waiting, SourceKind: "manual", Now: now,
	})
	require.NoError(t, err)
	activation := seed.Activations[0].Value
	activation.CreatedAt = now
	effect := seed.Effects[0].Value
	effect.Status = runstore.EffectStatusDispatching
	effect.LeaseOwner = "effect-worker"
	effect.LeaseToken = "effect-lease"
	leaseExpiry := now.Add(32 * time.Minute)
	effect.LeaseExpiresAt = &leaseExpiry
	repository := &engineRepositoryStub{activations: []runstore.Activation{activation}, effects: []runstore.Effect{effect}}
	engine, err := NewEngine(repository, materialLoaderStub{material: material}, EngineOptions{WorkerID: "worker"})
	require.NoError(t, err)
	require.NoError(t, engine.ProcessClaim(context.Background(), runstore.ClaimedRun{
		Run: run, Fence: runstore.TurnFence{WorkspaceID: run.WorkspaceID, RunID: run.ID, LeaseToken: run.LeaseToken},
	}, now.Add(31*time.Minute)))
	require.Len(t, repository.saved, 1)
	saved := repository.saved[0]
	require.Equal(t, runstore.RunStatusWaiting, saved.Status)
	require.Empty(t, saved.Effects)
	require.Equal(t, leaseExpiry, saved.RunnableAt)
}

func TestEngineDrainsCommittedDispatchBeforeFinalizingTerminalState(t *testing.T) {
	now := time.Date(2026, 7, 21, 6, 20, 0, 0, time.UTC)
	material := callbackMaterial(t)
	graph, err := runkernel.DecodeArtifact(material.ArtifactJSON)
	require.NoError(t, err)
	kernel, err := runkernel.NewKernel(graph)
	require.NoError(t, err)
	state, err := kernel.Start(runkernel.Values{"channel": "#workflow"})
	require.NoError(t, err)
	waiting, err := kernel.Step(state, now)
	require.NoError(t, err)
	terminal, err := kernel.Complete(waiting.State, runkernel.Completion{
		ActivationID: waiting.Intents[0].ActivationID, Port: "completed", Output: runkernel.Values{"reply": "done"},
	}, now)
	require.NoError(t, err)
	for !terminal.State.Status.Terminal() {
		next, stepErr := kernel.Step(terminal.State, now)
		require.NoError(t, stepErr)
		mergeTransition(&terminal, next)
	}
	stateJSON, err := json.Marshal(terminal.State)
	require.NoError(t, err)
	run := claimedRuntimeRun(now.Add(time.Second), stateJSON, 3)
	seed, err := ProjectTurn(ProjectTurnInput{
		Run: run, Fence: runstore.TurnFence{WorkspaceID: run.WorkspaceID, RunID: run.ID, LeaseToken: run.LeaseToken},
		Graph: graph, Transition: waiting, SourceKind: "manual", Now: now,
	})
	require.NoError(t, err)
	dispatching := seed.Effects[0].Value
	dispatching.Status = runstore.EffectStatusDispatching
	dispatching.LeaseOwner = "effect-worker"
	dispatching.LeaseToken = "effect-lease"
	leaseExpiry := now.Add(time.Minute)
	dispatching.LeaseExpiresAt = &leaseExpiry
	repository := &engineRepositoryStub{effects: []runstore.Effect{dispatching}}
	engine, err := NewEngine(repository, materialLoaderStub{material: material}, EngineOptions{WorkerID: "worker"})
	require.NoError(t, err)
	claim := runstore.ClaimedRun{
		Run: run, Fence: runstore.TurnFence{WorkspaceID: run.WorkspaceID, RunID: run.ID, LeaseToken: run.LeaseToken},
	}
	require.NoError(t, engine.ProcessClaim(context.Background(), claim, now.Add(time.Second)))
	require.Len(t, repository.saved, 1)
	draining := repository.saved[0]
	require.Equal(t, runstore.RunStatusWaiting, draining.Status)
	require.Nil(t, draining.CompletedAt)
	require.Empty(t, draining.Effects)

	completed := dispatching
	completed.Status = runstore.EffectStatusCompleted
	completed.LeaseOwner, completed.LeaseToken, completed.LeaseExpiresAt = "", "", nil
	completedAt := now.Add(2 * time.Second)
	completed.CompletedAt = &completedAt
	completed.ResultJSON, err = EncodeEffectResult(EffectResult{
		Port: "completed", Output: runkernel.Values{"reply": "done"}, MessageID: "msg",
		WakeAgentIDs: []string{"agt_callback"}, CompletedAt: &completedAt,
	})
	require.NoError(t, err)
	repository.effects = []runstore.Effect{completed}
	secondRun := run
	secondRun.Revision++
	secondRun.Status = runstore.RunStatusWaiting
	secondRun.StateJSON = draining.StateJSON
	secondRun.LeaseToken = "lease-2"
	claim = runstore.ClaimedRun{
		Run:   secondRun,
		Fence: runstore.TurnFence{WorkspaceID: secondRun.WorkspaceID, RunID: secondRun.ID, LeaseToken: secondRun.LeaseToken},
	}
	require.NoError(t, engine.ProcessClaim(context.Background(), claim, completedAt))
	require.Len(t, repository.saved, 2)
	finalized := repository.saved[1]
	require.Equal(t, runstore.RunStatusSucceeded, finalized.Status)
	require.NotNil(t, finalized.CompletedAt)
}

func TestEngineReductionBudgetLeavesCompletedReceiptImmediatelyRunnable(t *testing.T) {
	now := time.Date(2026, 7, 21, 6, 30, 0, 0, time.UTC)
	material := parallelCallbackMaterial(t)
	graph, err := runkernel.DecodeArtifact(material.ArtifactJSON)
	require.NoError(t, err)
	kernel, err := runkernel.NewKernel(graph)
	require.NoError(t, err)
	state, err := kernel.Start(runkernel.Values{"channel": "#workflow"})
	require.NoError(t, err)
	aggregate := runkernel.Transition{State: state, Intents: []runkernel.Intent{}, Events: []runkernel.Event{}}
	for len(runkernel.StableReadyIDs(aggregate.State)) != 0 {
		transition, stepErr := kernel.Step(aggregate.State, now)
		require.NoError(t, stepErr)
		mergeTransition(&aggregate, transition)
	}
	messageIntents := 0
	for _, intent := range aggregate.Intents {
		if intent.Kind == runkernel.IntentSendMessage {
			messageIntents++
		}
	}
	require.Equal(t, 2, messageIntents)
	require.Equal(t, runkernel.StatusWaiting, aggregate.State.Status)
	stateJSON, err := json.Marshal(aggregate.State)
	require.NoError(t, err)
	run := claimedRuntimeRun(now.Add(time.Second), stateJSON, 3)
	seed, err := ProjectTurn(ProjectTurnInput{
		Run: run, Fence: runstore.TurnFence{WorkspaceID: run.WorkspaceID, RunID: run.ID, LeaseToken: run.LeaseToken},
		Graph: graph, Transition: aggregate, SourceKind: "manual", Now: now,
	})
	require.NoError(t, err)
	activations := make([]runstore.Activation, 0, len(seed.Activations))
	for _, mutation := range seed.Activations {
		value := mutation.Value
		value.CreatedAt = now
		activations = append(activations, value)
	}
	effects := make([]runstore.Effect, 0, len(seed.Effects))
	for _, mutation := range seed.Effects {
		value := mutation.Value
		if value.Kind == EffectKindMessage {
			completedAt := now.Add(time.Second)
			value.Status = runstore.EffectStatusCompleted
			value.CompletedAt = &completedAt
			value.ResultJSON, err = EncodeEffectResult(EffectResult{
				Port: "completed", Output: runkernel.Values{"reply": value.ID}, MessageID: "msg_" + value.ID,
				WakeAgentIDs: []string{"agt_callback"}, CompletedAt: &completedAt,
			})
			require.NoError(t, err)
		}
		effects = append(effects, value)
	}
	repository := &engineRepositoryStub{activations: activations, effects: effects}
	engine, err := NewEngine(repository, materialLoaderStub{material: material}, EngineOptions{WorkerID: "worker", MaxReductions: 1})
	require.NoError(t, err)
	require.NoError(t, engine.ProcessClaim(context.Background(), runstore.ClaimedRun{
		Run: run, Fence: runstore.TurnFence{WorkspaceID: run.WorkspaceID, RunID: run.ID, LeaseToken: run.LeaseToken},
	}, now.Add(time.Second)))
	require.Len(t, repository.saved, 1)
	saved := repository.saved[0]
	require.Equal(t, now.Add(time.Second), saved.RunnableAt)
	var persisted runkernel.State
	require.NoError(t, json.Unmarshal(saved.StateJSON, &persisted))
	waitingCallbacks := 0
	for _, activation := range persisted.Activations {
		if activation.Status == runkernel.ActivationWaiting && (activation.NodeID == "left_agent" || activation.NodeID == "right_agent") {
			waitingCallbacks++
		}
	}
	require.Equal(t, 1, waitingCallbacks, "one completed receipt must remain for the next bounded turn")
	require.Empty(t, runkernel.StableReadyIDs(&persisted), "the remaining receipt, not a ready activation, must drive immediate scheduling")
	var remainingLogicalKey string
	for _, activation := range persisted.Activations {
		if activation.Status == runkernel.ActivationWaiting && (activation.NodeID == "left_agent" || activation.NodeID == "right_agent") {
			remainingLogicalKey = activation.ID
		}
	}
	require.NotEmpty(t, remainingLogicalKey)
	for _, mutation := range saved.Activations {
		require.NotEqual(t, remainingLogicalKey, mutation.Value.LogicalKey,
			"a sibling callback waiting without a repeated intent must retain its original deadline without a clearing update")
	}
}

func TestEngineCollectionCapProbesExactlyOneBeyondLimit(t *testing.T) {
	engine, err := NewEngine(&engineRepositoryStub{}, materialLoaderStub{}, EngineOptions{WorkerID: "worker"})
	require.NoError(t, err)
	run := runstore.Run{ID: "wri", WorkspaceID: "wsp"}
	exactActivations := make([]runstore.Activation, maxRuntimeChildren)
	exactEffects := make([]runstore.Effect, maxRuntimeChildren)
	for index := 0; index < maxRuntimeChildren; index++ {
		exactActivations[index].ID = fmt.Sprintf("wra_%05d", index)
		exactEffects[index].ID = fmt.Sprintf("wfx_%05d", index)
	}
	repository := engine.repository.(*engineRepositoryStub)
	repository.activations = exactActivations
	repository.effects = exactEffects
	loadedActivations, err := engine.listAllActivations(context.Background(), run)
	require.NoError(t, err)
	require.Len(t, loadedActivations, maxRuntimeChildren)
	loadedEffects, err := engine.listAllEffects(context.Background(), run)
	require.NoError(t, err)
	require.Len(t, loadedEffects, maxRuntimeChildren)

	repository.activations = append(repository.activations, runstore.Activation{ID: "wra_10000"})
	_, err = engine.listAllActivations(context.Background(), run)
	require.ErrorIs(t, err, errRuntimeCollectionLimit)
	repository.activations = nil
	repository.effects = append(repository.effects, runstore.Effect{ID: "wfx_10000"})
	_, err = engine.listAllEffects(context.Background(), run)
	require.ErrorIs(t, err, errRuntimeCollectionLimit)
}

func TestEngineTerminalsRunWhenRuntimeCollectionExceedsLimit(t *testing.T) {
	now := time.Date(2026, 7, 21, 6, 45, 0, 0, time.UTC)
	material := callbackMaterial(t)
	for _, testCase := range []struct {
		name string
		seed func(*engineRepositoryStub)
	}{
		{name: "activations", seed: func(repository *engineRepositoryStub) {
			repository.activations = make([]runstore.Activation, maxRuntimeChildren+1)
			for index := range repository.activations {
				repository.activations[index].ID = fmt.Sprintf("wra_%05d", index)
			}
		}},
		{name: "effects", seed: func(repository *engineRepositoryStub) {
			repository.effects = make([]runstore.Effect, maxRuntimeChildren+1)
			for index := range repository.effects {
				repository.effects[index].ID = fmt.Sprintf("wfx_%05d", index)
			}
		}},
	} {
		t.Run(testCase.name, func(t *testing.T) {
			repository := &engineRepositoryStub{}
			testCase.seed(repository)
			engine, err := NewEngine(repository, materialLoaderStub{material: material}, EngineOptions{WorkerID: "worker"})
			require.NoError(t, err)
			run := claimedRuntimeRun(now, []byte(`{}`), 2)
			err = engine.ProcessClaim(context.Background(), runstore.ClaimedRun{
				Run: run, Fence: runstore.TurnFence{WorkspaceID: run.WorkspaceID, RunID: run.ID, LeaseToken: run.LeaseToken},
			}, now)
			require.NoError(t, err)
			require.Len(t, repository.failed, 1)
			require.Equal(t, runtimeCollectionLimitErrorCode, repository.failed[0].ErrorCode)
		})
	}
}

func TestEnginePersistsMaterialFailureAsTerminalRun(t *testing.T) {
	now := time.Date(2026, 7, 21, 6, 0, 0, 0, time.UTC)
	material := callbackMaterial(t)
	material.PlanVersionID = "wrong"
	run := claimedRuntimeRun(now, []byte(`{}`), 2)
	repository := &engineRepositoryStub{}
	engine, err := NewEngine(repository, materialLoaderStub{material: material}, EngineOptions{WorkerID: "worker"})
	require.NoError(t, err)
	require.NoError(t, engine.ProcessClaim(context.Background(), runstore.ClaimedRun{
		Run: run, Fence: runstore.TurnFence{WorkspaceID: run.WorkspaceID, RunID: run.ID, LeaseToken: run.LeaseToken},
	}, now))
	require.Len(t, repository.failed, 1)
	require.Equal(t, "run.plan_version_invalid", repository.failed[0].ErrorCode)
	var failed runkernel.State
	require.NoError(t, json.Unmarshal(repository.failed[0].StateJSON, &failed))
	require.Equal(t, runkernel.StatusFailed, failed.Status)
	require.NotNil(t, failed.Input)
	require.NotNil(t, failed.Variables)
	require.NotNil(t, failed.Activations)
	require.NotNil(t, failed.Scopes)
}

func TestEngineRetriesUnknownMaterialLoaderFailureWithoutChangingRun(t *testing.T) {
	now := time.Date(2026, 7, 21, 6, 5, 0, 0, time.UTC)
	run := claimedRuntimeRun(now, []byte(`{}`), 2)
	repository := &engineRepositoryStub{}
	transient := fmt.Errorf("temporary database read timeout")
	engine, err := NewEngine(repository, materialLoaderStub{err: transient}, EngineOptions{WorkerID: "worker"})
	require.NoError(t, err)
	err = engine.ProcessClaim(context.Background(), runstore.ClaimedRun{
		Run: run, Fence: runstore.TurnFence{WorkspaceID: run.WorkspaceID, RunID: run.ID, LeaseToken: run.LeaseToken},
	}, now)
	require.ErrorIs(t, err, transient)
	require.Empty(t, repository.saved)
	require.Empty(t, repository.failed)
}

func TestEnginePersistsProvenPermanentMaterialLoaderFailure(t *testing.T) {
	now := time.Date(2026, 7, 21, 6, 10, 0, 0, time.UTC)
	run := claimedRuntimeRun(now, []byte(`{}`), 2)
	repository := &engineRepositoryStub{}
	permanent := permanentMaterialError("exact PlanVersion is gone", nil)
	engine, err := NewEngine(repository, materialLoaderStub{err: permanent}, EngineOptions{WorkerID: "worker"})
	require.NoError(t, err)
	require.NoError(t, engine.ProcessClaim(context.Background(), runstore.ClaimedRun{
		Run: run, Fence: runstore.TurnFence{WorkspaceID: run.WorkspaceID, RunID: run.ID, LeaseToken: run.LeaseToken},
	}, now))
	require.Len(t, repository.failed, 1)
	require.Equal(t, "run.plan_version_unavailable", repository.failed[0].ErrorCode)
}

func runFirstTaskWatchBarrierTurn(
	t *testing.T,
	now time.Time,
	joinMode string,
) (Material, runstore.Run, runstore.SaveTurnParams) {
	t.Helper()
	material := taskWatchBarrierMaterial(t, joinMode)
	run := claimedRuntimeRun(now, []byte(`{}`), 2)
	run.InputJSON = taskWatchBarrierInputJSON()
	repository := &engineRepositoryStub{}
	engine, err := NewEngine(
		repository,
		materialLoaderStub{material: material},
		EngineOptions{WorkerID: "worker"},
	)
	require.NoError(t, err)
	require.NoError(t, engine.ProcessClaim(context.Background(), claimedRun(run), now))
	require.Len(t, repository.saved, 1)
	return material, run, repository.saved[0]
}

func resumedTaskWatchBarrierRun(
	prior runstore.Run,
	saved runstore.SaveTurnParams,
	now time.Time,
) runstore.Run {
	run := prior
	run.Status = saved.Status
	run.StateJSON = append(json.RawMessage(nil), saved.StateJSON...)
	run.Revision++
	run.RunnableAt = now
	run.LeaseOwner = "worker"
	run.LeaseToken = fmt.Sprintf("lease-%d", run.Revision)
	leaseExpiry := now.Add(time.Minute)
	run.LeaseExpires = &leaseExpiry
	return run
}

func claimedRun(run runstore.Run) runstore.ClaimedRun {
	return runstore.ClaimedRun{
		Run: run,
		Fence: runstore.TurnFence{
			WorkspaceID: run.WorkspaceID, RunID: run.ID, LeaseToken: run.LeaseToken,
		},
	}
}

func decodeSavedRunState(t *testing.T, payload []byte) *runkernel.State {
	t.Helper()
	state, initialized, err := decodeRunState(payload)
	require.NoError(t, err)
	require.True(t, initialized)
	return state
}

func requireRunActivationByNode(
	t *testing.T,
	state *runkernel.State,
	nodeID string,
) *runkernel.Activation {
	t.Helper()
	require.NotNil(t, state)
	for index := range state.Activations {
		if state.Activations[index].NodeID == nodeID {
			return &state.Activations[index]
		}
	}
	t.Fatalf("Run state has no activation for node %q", nodeID)
	return nil
}

func activationMutationValues(
	mutations []runstore.ActivationMutation,
) []runstore.Activation {
	values := make([]runstore.Activation, 0, len(mutations))
	for _, mutation := range mutations {
		values = append(values, mutation.Value)
	}
	return values
}

func insertedEffectMutations(
	mutations []runstore.EffectMutation,
) []runstore.Effect {
	values := make([]runstore.Effect, 0, len(mutations))
	for _, mutation := range mutations {
		if mutation.Operation == runstore.MutationInsert {
			values = append(values, mutation.Value)
		}
	}
	return values
}

func insertedTaskWatchEffects(
	t *testing.T,
	mutations []runstore.EffectMutation,
) []runstore.Effect {
	t.Helper()
	values := make([]runstore.Effect, 0, len(mutations))
	for _, effect := range insertedEffectMutations(mutations) {
		request, err := DecodeEffectRequest(effect.RequestJSON)
		require.NoError(t, err)
		if request.Intent.Kind == runkernel.IntentSendMessage &&
			request.Intent.Callback && request.Intent.TaskWatchHandoff != nil {
			values = append(values, effect)
		}
	}
	return values
}

func replaceEffectByID(
	t *testing.T,
	effects []runstore.Effect,
	replacement runstore.Effect,
) {
	t.Helper()
	for index := range effects {
		if effects[index].ID == replacement.ID {
			effects[index] = replacement
			return
		}
	}
	t.Fatalf("persisted effect %q was not found", replacement.ID)
}

func effectMutationsWithStatus(
	mutations []runstore.EffectMutation,
	status string,
) []runstore.EffectMutation {
	values := make([]runstore.EffectMutation, 0, len(mutations))
	for _, mutation := range mutations {
		if mutation.Value.Status == status {
			values = append(values, mutation)
		}
	}
	return values
}

func indexOfString(values []string, target string) int {
	for index, value := range values {
		if value == target {
			return index
		}
	}
	return -1
}

func stepToRuntimeTaskWatchIntent(
	t *testing.T,
	kernel *runkernel.Kernel,
	state *runkernel.State,
	now time.Time,
) (runkernel.Intent, *runkernel.State) {
	t.Helper()
	for step := 0; step < 32; step++ {
		transition, err := kernel.Step(state, now)
		require.NoError(t, err)
		state = transition.State
		for _, intent := range transition.Intents {
			if intent.Kind == runkernel.IntentSendMessage && intent.Callback && intent.TaskWatchHandoff != nil {
				return intent, state
			}
		}
	}
	t.Fatalf("no fixed-Task callback intent became ready")
	return runkernel.Intent{}, nil
}

func taskWatchBarrierScope() domain.MessageScopeValue {
	return domain.MessageScopeValue{
		Kind: domain.MessageScopeKindTask, ScopeID: "tsk_shared",
	}
}

func taskWatchBarrierInputValues() runkernel.Values {
	return runkernel.Values{
		"task": map[string]any{
			"kind": string(domain.MessageScopeKindTask), "scopeId": "tsk_shared",
		},
		"agent_a": "@agent-a",
		"agent_b": "@agent-b",
	}
}

func taskWatchBarrierInputJSON() json.RawMessage {
	payload, err := json.Marshal(taskWatchBarrierInputValues())
	if err != nil {
		panic(err)
	}
	return payload
}

func taskWatchBarrierCallbackOutput(value string) runkernel.Values {
	return runkernel.Values{
		"value": value,
		"workScope": map[string]any{
			"kind": string(domain.MessageScopeKindTask), "scopeId": "tsk_shared",
		},
	}
}

func taskWatchBarrierMaterial(t *testing.T, joinMode string) Material {
	t.Helper()
	mergeRef := "branches.left.value"
	if joinMode == "any" {
		mergeRef = "winner.value.value"
	}
	source := fmt.Sprintf(`apiVersion: loop.plan/v1
kind: PlanVersion
input:
  task: {type: scope, required: true}
  agent_a: {type: agent, required: true}
  agent_b: {type: agent, required: true}
graph:
  entry: {nodeId: split, port: in}
  nodes:
  - id: split
    type: fork
    fork: {branches: [left, right]}
  - id: a
    type: send_message
    input:
      task: {type: scope, required: true}
      agent: {type: agent, required: true}
    sendMessage:
      scope: {ref: input.task}
      responder: {ref: input.agent}
      text: branch A
      completion: callback
      workScope: {ref: input.task}
      callbackOutput: {value: {type: string, required: true}}
  - id: b
    type: send_message
    input:
      task: {type: scope, required: true}
      agent: {type: agent, required: true}
    sendMessage:
      scope: {ref: input.task}
      responder: {ref: input.agent}
      text: branch B
      completion: callback
      workScope: {ref: input.task}
      callbackOutput: {value: {type: string, required: true}}
  - id: gather
    type: join
    input:
      value: {type: string, required: true}
    join:
      forkId: split
      mode: %s
      result:
        value: {type: string, required: true}
      merge:
        value: {ref: %s}
  - id: success
    type: end
    end: {outcome: succeeded}
  - id: failure
    type: end
    end: {outcome: failed, code: parallel_failed}
  edges:
  - id: start_left
    source: {nodeId: split, port: left}
    target: {nodeId: a, port: in}
    bindings:
      task: {ref: plan.input.task}
      agent: {ref: plan.input.agent_a}
  - id: start_right
    source: {nodeId: split, port: right}
    target: {nodeId: b, port: in}
    bindings:
      task: {ref: plan.input.task}
      agent: {ref: plan.input.agent_b}
  - id: left_completed
    source: {nodeId: a, port: completed}
    target: {nodeId: gather, port: left}
    bindings:
      value: {ref: output.value}
  - id: left_failed
    source: {nodeId: a, port: failed}
    target: {nodeId: gather, port: left}
    bindings:
      value: {value: ""}
  - id: right_completed
    source: {nodeId: b, port: completed}
    target: {nodeId: gather, port: right}
    bindings:
      value: {ref: output.value}
  - id: right_failed
    source: {nodeId: b, port: failed}
    target: {nodeId: gather, port: right}
    bindings:
      value: {value: ""}
  - id: joined
    source: {nodeId: gather, port: completed}
    target: {nodeId: success, port: in}
  - id: join_failed
    source: {nodeId: gather, port: failed}
    target: {nodeId: failure, port: in}
`, joinMode, mergeRef)
	artifact, diagnostics, err := plan.Compile([]byte(source))
	require.NoError(t, err)
	require.False(t, plan.HasErrors(diagnostics), "diagnostics: %#v", diagnostics)
	wire, err := artifact.DeterministicJSON()
	require.NoError(t, err)
	return Material{
		WorkspaceID: "wsp", PlanID: "pln", PlanVersionID: "plv",
		CompilerVersion: artifact.CompilerVersion, Digest: artifact.Digest,
		ArtifactJSON: wire, SourceKind: "manual",
	}
}

func callbackMaterial(t *testing.T) Material {
	t.Helper()
	raw, err := os.ReadFile("../plan/testdata/ir/send_message_callback.json")
	require.NoError(t, err)
	artifact, err := plan.DecodeArtifact(raw)
	require.NoError(t, err)
	return Material{
		WorkspaceID: "wsp", PlanID: "pln", PlanVersionID: "plv",
		CompilerVersion: artifact.CompilerVersion, Digest: artifact.Digest, ArtifactJSON: raw, SourceKind: "manual",
	}
}

func parallelCallbackMaterial(t *testing.T) Material {
	t.Helper()
	const source = `apiVersion: loop.plan/v1
kind: PlanVersion
input:
  channel: {type: channel, required: true}
output:
  reply: {type: string, required: false}
graph:
  entry:
    nodeId: fanout
    port: in
    input:
      channel: {ref: plan.input.channel}
  nodes:
  - id: fanout
    type: fork
    input:
      channel: {type: channel, required: true}
    fork: {branches: [left, right]}
  - id: left_agent
    type: send_message
    input:
      channel: {type: channel, required: true}
    sendMessage:
      target: "{{ input.channel }}"
      text: "@left answer"
      completion: callback
      callbackOutput:
        reply: {type: string, required: true}
      retry: {maxAttempts: 1, backoff: 1s}
  - id: right_agent
    type: send_message
    input:
      channel: {type: channel, required: true}
    sendMessage:
      target: "{{ input.channel }}"
      text: "@right answer"
      completion: callback
      callbackOutput:
        reply: {type: string, required: true}
      retry: {maxAttempts: 1, backoff: 1s}
  - id: gather
    type: join
    input:
      reply: {type: string, required: false}
    join:
      forkId: fanout
      mode: all
      result:
        reply: {type: string, required: false}
      merge:
        reply: {ref: branches.left.reply}
  - id: success
    type: end
    end: {outcome: succeeded}
  - id: failure
    type: end
    end: {outcome: failed, code: parallel_failed}
  edges:
  - id: start_left
    source: {nodeId: fanout, port: left}
    target: {nodeId: left_agent, port: in}
    bindings:
      channel: {ref: input.channel}
  - id: start_right
    source: {nodeId: fanout, port: right}
    target: {nodeId: right_agent, port: in}
    bindings:
      channel: {ref: input.channel}
  - id: left_completed
    source: {nodeId: left_agent, port: completed}
    target: {nodeId: gather, port: left}
    bindings:
      reply: {ref: output.reply}
  - id: left_failed
    source: {nodeId: left_agent, port: failed}
    target: {nodeId: gather, port: left}
    bindings:
      reply: {value: ""}
  - id: right_completed
    source: {nodeId: right_agent, port: completed}
    target: {nodeId: gather, port: right}
    bindings:
      reply: {ref: output.reply}
  - id: right_failed
    source: {nodeId: right_agent, port: failed}
    target: {nodeId: gather, port: right}
    bindings:
      reply: {value: ""}
  - id: joined
    source: {nodeId: gather, port: completed}
    target: {nodeId: success, port: in}
    bindings:
      reply: {ref: output.reply}
  - id: join_failed
    source: {nodeId: gather, port: failed}
    target: {nodeId: failure, port: in}
`
	artifact, diagnostics, err := plan.Compile([]byte(source))
	require.NoError(t, err)
	require.False(t, plan.HasErrors(diagnostics), "diagnostics: %#v", diagnostics)
	wire, err := artifact.DeterministicJSON()
	require.NoError(t, err)
	return Material{
		WorkspaceID: "wsp", PlanID: "pln", PlanVersionID: "plv",
		CompilerVersion: artifact.CompilerVersion, Digest: artifact.Digest, ArtifactJSON: wire, SourceKind: "manual",
	}
}

func claimedRuntimeRun(now time.Time, state []byte, revision int64) runstore.Run {
	leaseExpiry := now.Add(time.Minute)
	started := now
	return runstore.Run{
		ID: "wri_run", WorkspaceID: "wsp", InvocationID: "wiv", PlanID: "pln", PlanVersionID: "plv",
		Status: runstore.RunStatusRunning, InputJSON: []byte(`{"channel":"#workflow"}`), InputDigest: "digest",
		StateJSON: state, Identity: invocationIdentity(), StartedAt: &started,
		Revision: revision, LastEventSeq: 1, RunnableAt: now,
		LeaseOwner: "worker", LeaseToken: "lease", LeaseExpires: &leaseExpiry,
	}
}

func eventTypes(values []runstore.EventDraft) []string {
	result := make([]string, len(values))
	for index := range values {
		result[index] = values[index].Type
	}
	return result
}

func filterActivationsAfter(values []runstore.Activation, after string, limit int) []runstore.Activation {
	result := make([]runstore.Activation, 0)
	for _, value := range values {
		if value.ID > after {
			result = append(result, value)
		}
		if len(result) == limit {
			break
		}
	}
	return result
}

func filterEffectsAfter(values []runstore.Effect, after string, limit int) []runstore.Effect {
	result := make([]runstore.Effect, 0)
	for _, value := range values {
		if value.ID > after {
			result = append(result, value)
		}
		if len(result) == limit {
			break
		}
	}
	return result
}