package orchestrator

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

	"github.com/tidbcloud/AgentTeam/internal/core/domain"
	workflowrunstore "github.com/tidbcloud/AgentTeam/internal/store/workflow/run"
	runkernel "github.com/tidbcloud/AgentTeam/internal/workflow/run"
	"github.com/tidbcloud/AgentTeam/internal/workflow/runartifact"
)

const (
	defaultEffectClaimLimit = 64
	defaultEffectLeaseTTL   = 30 * time.Second
)

// ArtifactContentResolver is implemented by the production Run store. It
// deliberately accepts the complete typed reference, never a caller-supplied
// artifact ID, so effect delivery cannot bypass Run/workspace binding.
type ArtifactContentResolver interface {
	ResolveArtifactContent(context.Context, runartifact.Ref) ([]byte, error)
}

type EffectRepository interface {
	ListPendingEffects(context.Context, workflowrunstore.ListPendingEffectsParams) ([]workflowrunstore.EffectCandidate, error)
	ClaimEffect(context.Context, workflowrunstore.ClaimEffectParams) (workflowrunstore.Effect, bool, error)
	ClaimCallbackEffectCapacity(context.Context, workflowrunstore.ClaimCallbackEffectCapacityParams) (workflowrunstore.ClaimCallbackEffectCapacityResult, error)
	// SelectCallbackResponderCapacity is the responder.selectOne capacity gate:
	// it atomically selects the highest-priority free candidate from the frozen
	// ordered pool or durably parks the effect as waiting_for_responder.
	SelectCallbackResponderCapacity(context.Context, workflowrunstore.SelectCallbackResponderCapacityParams) (workflowrunstore.SelectCallbackResponderCapacityResult, error)
	// PromoteCallbackResponderWaiters wakes the oldest eligible
	// waiting_for_responder waiters whose candidate pool now has a free
	// candidate, returning the promoted effects the caller must deliver.
	PromoteCallbackResponderWaiters(context.Context, string, workflowrunstore.PromoteCallbackResponderWaitersParams, int) ([]workflowrunstore.SelectCallbackResponderCapacityResult, error)
	ReconcileAgentExecutionLeases(context.Context, string, int) (int, error)
	FreezeApprovalAssignee(context.Context, string, string, string, string, string) error
	ResolveEffect(context.Context, workflowrunstore.ResolveEffectParams) (workflowrunstore.Effect, []workflowrunstore.Event, error)
	LoadRun(context.Context, string, string) (workflowrunstore.Run, error)
}

// ApprovalAssigneeResolver turns an author-authored assignee reference into
// one exact current Human member before the Workflow App notification is
// created. The resolved member id is persisted in the effect receipt and is
// the sole Human authorized to decide it.
type ApprovalAssigneeResolver interface {
	ResolveWorkflowApprovalAssignee(context.Context, string, string) (string, error)
}

// CallbackResponderResolver resolves the resolved callback Agent for a
// rendered send_message.callback request BEFORE delivery, using the same
// trusted AgentRef/destination-visibility chain as the ordinary message
// transaction. The resolved Agent ID is frozen on the effect and gates the
// workspace+Agent execution capacity.
type CallbackResponderResolver interface {
	ResolveCallbackResponder(context.Context, MessageRequest) (string, error)
}

type EffectDispatcherOptions struct {
	WorkerID         string
	WorkspaceGroupID string
	ClaimLimit       int
	LeaseTTL         time.Duration
	Now              func() time.Time
	Responders       CallbackResponderResolver
	// Randomizer is the injectable unbiased source for the "random" selectOne
	// strategy, forwarded to the store capacity gate. Nil defaults to the store's
	// crypto-backed source so production is never predictable; a test injects a
	// seeded Randomizer to drive a deterministic draw through the real store.
	Randomizer workflowrunstore.Randomizer
	// AfterMessageCommitted is a production-shared fault seam invoked at the
	// narrow window AFTER the native message transaction committed + bound the
	// effect's MessageID but BEFORE the receipt is resolved awaiting. Production
	// wires nil (no-op). A deterministic test returns a bounded error to simulate
	// a process crash at exactly that window: DispatchOnce reports the failure
	// WITHOUT writing a failed receipt, so the effect stays dispatching with a
	// frozen responder + bound message, and a fresh dispatcher's normal recovery
	// path must reuse the frozen responder (never re-draw). The seam is bounded
	// to one call site and carries no test-only state into production.
	AfterMessageCommitted func() error
}

type EffectDispatcher struct {
	repository            EffectRepository
	messages              MessageWriter
	approvals             ApprovalWriter
	assignees             ApprovalAssigneeResolver
	responders            CallbackResponderResolver
	workerID              string
	workspaceGroupID      string
	claimLimit            int
	leaseTTL              time.Duration
	now                   func() time.Time
	randomizer            workflowrunstore.Randomizer
	afterMessageCommitted func() error
}

func NewEffectDispatcher(
	repository EffectRepository,
	messages MessageWriter,
	approvals ApprovalWriter,
	assignees ApprovalAssigneeResolver,
	options EffectDispatcherOptions,
) (*EffectDispatcher, error) {
	if repository == nil {
		return nil, fmt.Errorf("workflow Run effect repository is required")
	}
	options.WorkerID = strings.TrimSpace(options.WorkerID)
	if options.WorkerID == "" || len(options.WorkerID) > 128 {
		return nil, fmt.Errorf("workflow Run effect workerId is required and must be bounded")
	}
	if options.ClaimLimit == 0 {
		options.ClaimLimit = defaultEffectClaimLimit
	}
	if options.ClaimLimit < 1 || options.ClaimLimit > runtimeCollectionPage {
		return nil, fmt.Errorf("workflow Run effect claim limit must be between 1 and %d", runtimeCollectionPage)
	}
	if options.LeaseTTL == 0 {
		options.LeaseTTL = defaultEffectLeaseTTL
	}
	if options.LeaseTTL <= 0 {
		return nil, fmt.Errorf("workflow Run effect lease TTL must be positive")
	}
	if options.Now == nil {
		options.Now = time.Now
	}
	return &EffectDispatcher{
		repository: repository, messages: messages, approvals: approvals, assignees: assignees, responders: options.Responders,
		workerID: options.WorkerID, workspaceGroupID: strings.TrimSpace(options.WorkspaceGroupID),
		claimLimit: options.ClaimLimit, leaseTTL: options.LeaseTTL, now: options.Now, randomizer: options.Randomizer,
		afterMessageCommitted: options.AfterMessageCommitted,
	}, nil
}

type DispatchOutcome struct {
	Candidates int
	Claimed    int
	Resolved   int
	Failed     int
}

func (dispatcher *EffectDispatcher) DispatchOnce(ctx context.Context) (DispatchOutcome, error) {
	if dispatcher == nil || dispatcher.repository == nil {
		return DispatchOutcome{}, fmt.Errorf("workflow Run effect dispatcher is not configured")
	}
	scanAt := dispatcher.clockNow()
	candidates, err := dispatcher.repository.ListPendingEffects(ctx, workflowrunstore.ListPendingEffectsParams{
		WorkspaceGroupID: dispatcher.workspaceGroupID, Now: scanAt, Limit: dispatcher.claimLimit,
	})
	if err != nil {
		return DispatchOutcome{}, err
	}
	outcome := DispatchOutcome{Candidates: len(candidates)}
	var failures []error
	for _, candidate := range candidates {
		// A prior network dispatch may consume most or all of a lease TTL. Each
		// candidate therefore gets a lease rooted at its actual claim time, not
		// at the beginning of this bounded scan.
		claimAt := dispatcher.clockNow()
		effect, claimed, err := dispatcher.repository.ClaimEffect(ctx, workflowrunstore.ClaimEffectParams{
			Candidate: candidate, WorkerID: dispatcher.workerID, Now: claimAt, LeaseTTL: dispatcher.leaseTTL,
		})
		if err != nil {
			outcome.Failed++
			failures = append(failures, err)
			continue
		}
		if !claimed {
			continue
		}
		outcome.Claimed++
		if err := dispatcher.dispatchClaim(ctx, effect); err != nil {
			outcome.Failed++
			failures = append(failures, fmt.Errorf("dispatch workflow Run effect %s: %w", effect.ID, err))
			continue
		}
		outcome.Resolved++
	}
	return outcome, errors.Join(failures...)
}

func (dispatcher *EffectDispatcher) clockNow() time.Time {
	if dispatcher == nil || dispatcher.now == nil {
		return time.Time{}
	}
	return dispatcher.now().UTC().Truncate(time.Microsecond)
}

// ReconcileOnce releases provably-terminal execution leases and wakes the
// oldest eligible responder.selectOne waiters whose candidate pool now has a
// free candidate (bounded, per workspace group). Promoted waiters are already
// dispatching under this worker's token with a frozen responder, so they are
// delivered immediately by this pass; the normal pending scan does not match
// dispatching effects. It never releases live awaiting/pending/dispatching
// owners and never pokes another Run.
func (dispatcher *EffectDispatcher) ReconcileOnce(ctx context.Context) error {
	if dispatcher == nil || dispatcher.repository == nil {
		return nil
	}
	var failures []error
	if _, err := dispatcher.repository.ReconcileAgentExecutionLeases(ctx, dispatcher.workspaceGroupID, runtimeCollectionPage); err != nil {
		failures = append(failures, err)
	}
	promoted, err := dispatcher.repository.PromoteCallbackResponderWaiters(ctx, dispatcher.workspaceGroupID,
		workflowrunstore.PromoteCallbackResponderWaitersParams{WorkerID: dispatcher.workerID, Now: dispatcher.clockNow(), LeaseTTL: dispatcher.leaseTTL},
		runtimeCollectionPage)
	if err != nil {
		failures = append(failures, err)
	}
	for _, result := range promoted {
		if !result.Claimed || result.Waiting {
			continue
		}
		if err := dispatcher.dispatchClaim(ctx, result.Effect); err != nil {
			failures = append(failures, err)
		}
	}
	return errors.Join(failures...)
}

func (dispatcher *EffectDispatcher) dispatchClaim(ctx context.Context, effect workflowrunstore.Effect) error {
	request, err := DecodeEffectRequest(effect.RequestJSON)
	if err != nil {
		return dispatcher.resolveFailure(ctx, effect, request, "effect.request_invalid", err, false)
	}
	switch request.Intent.Kind {
	case runkernel.IntentSendMessage:
		return dispatcher.dispatchMessage(ctx, effect, request)
	case runkernel.IntentWait:
		now := dispatcher.clockNow()
		if effectDeadlineExpired(effect, request, now) {
			return dispatcher.resolveTimedOut(ctx, effect, request, now)
		}
		return dispatcher.resolvePort(ctx, effect, EffectResult{Port: "completed", Output: runkernel.Values{}},
			workflowrunstore.EffectStatusCompleted, effectEvent("wait_completed", effect, request, nil), now, now)
	case runkernel.IntentApproval:
		return dispatcher.dispatchApproval(ctx, effect, request)
	default:
		return dispatcher.resolveFailure(ctx, effect, request, "effect.kind_unsupported", fmt.Errorf("unsupported effect kind %q", request.Intent.Kind), false)
	}
}

func (dispatcher *EffectDispatcher) dispatchMessage(ctx context.Context, effect workflowrunstore.Effect, request EffectRequest) error {
	// A bound MessageID proves the external transaction committed even if the
	// receipt was lost before ResolveEffect; recovery reconstructs the frozen
	// durable receipt instead of projecting a false message-less failure.
	if dispatcher.messages == nil || request.Message == nil {
		return dispatcher.resolveFailure(ctx, effect, request, "message.writer_unavailable", fmt.Errorf("workflow Run message writer is not configured"), false)
	}
	if err := dispatcher.materializeMessageContent(ctx, request.Message); err != nil {
		return dispatcher.resolveFailure(ctx, effect, request, "message.content_unavailable", err, true)
	}
	boundMessageID := strings.TrimSpace(effect.MessageID)
	var result MessageResult
	var err error
	frozenResponderAgentID := ""
	if boundMessageID != "" {
		recoverer, ok := dispatcher.messages.(MessageReceiptRecoverer)
		if !ok {
			return dispatcher.resolveFailure(
				ctx,
				effect,
				request,
				"message.receipt_recovery_unavailable",
				fmt.Errorf("workflow Run message receipt recoverer is not configured"),
				true,
			)
		}
		// The message transaction already committed; re-running the ordinary
		// writer could fail after mutable state changed. Recover the frozen
		// receipt directly from the durable fanout record instead.
		result, err = recoverer.RecoverRunMessageReceipt(ctx, *request.Message, boundMessageID)
	} else {
		if request.Intent.Callback {
			// Workspace+Agent capacity is gated BEFORE the message transaction so a
			// queued callback never produces a visible contribution out of order; the
			// frozen responder is resolved + atomically persisted with the capacity
			// acquisition and is never re-resolved from a mutable AgentRef.
			gated, frozenAgentID, gateErr := dispatcher.gateCallbackDelivery(ctx, effect, request)
			if gateErr != nil {
				return dispatcher.resolveFailure(ctx, effect, request, "callback.responder_unresolvable", gateErr, false)
			}
			if !gated {
				return nil
			}
			frozenResponderAgentID = frozenAgentID
			// Trusted fence: the message tx rolls back with zero side effects if
			// the resolved responder drifts from this frozen ID.
			runtimeMessage := *request.Message
			runtimeMessage.ExpectedCallbackResponderAgentID = frozenResponderAgentID
			// responder.selectOne freezes one winner at the capacity gate. The
			// ordinary message transaction resolves its single responder from the
			// responder ref (or a direct mention); it has no notion of a candidate
			// pool. Materialize the frozen winner as the responder ref on the
			// runtime message and drop the resolved pool so the real writer
			// delivers to exactly the winner it must, still fenced by
			// ExpectedCallbackResponderAgentID above.
			if len(runtimeMessage.ResponderCandidates) > 0 {
				runtimeMessage.ResponderAgentRef = frozenResponderAgentID
				runtimeMessage.ResponderCandidates = nil
			}
			result, err = dispatcher.messages.WriteRunMessage(ctx, runtimeMessage)
		} else {
			result, err = dispatcher.messages.WriteRunMessage(ctx, *request.Message)
		}
	}
	if err != nil {
		return dispatcher.resolveFailure(ctx, effect, request, "message.delivery_failed", err, true)
	}
	if dispatcher.afterMessageCommitted != nil {
		// Deterministic fault seam: the native message tx has committed and bound
		// the effect's MessageID, but the receipt is not yet resolved. A test
		// returns a bounded error here so DispatchOnce reports the failure WITHOUT
		// writing a failed receipt — the effect stays dispatching + bound, and
		// recovery reuses the frozen responder.
		if err := dispatcher.afterMessageCommitted(); err != nil {
			return err
		}
	}
	completedAt := dispatcher.clockNow()
	result = result.Normalize()
	receipt, err := validatedMessageReceipt(request, result, boundMessageID)
	if err != nil {
		return dispatcher.resolveFailure(ctx, effect, request, "message.delivery_failed", err, true)
	}
	if err := dispatcher.validateCallbackResponderReceipt(ctx, effect, request, receipt, frozenResponderAgentID); err != nil {
		return err
	}
	if request.Intent.Callback {
		return dispatcher.resolveCallbackAwaiting(ctx, effect, request, result, receipt, completedAt)
	}
	receipt.Port = "completed"
	receipt.Output = runkernel.Values{}
	if request.Intent.StartThread {
		receipt.Output = runkernel.Values{
			"messageId": result.MessageID,
			"threadScope": map[string]any{
				"kind":    string(result.ThreadScope.Kind),
				"scopeId": result.ThreadScope.ScopeID,
			},
		}
	}
	return dispatcher.resolvePort(ctx, effect, receipt, workflowrunstore.EffectStatusCompleted,
		effectEvent("message_sent", effect, request, map[string]any{
			"messageId": result.MessageID, "deliveryScope": result.DeliveryScope, "threadScope": result.ThreadScope,
		}), completedAt, completedAt)
}

// resolveCallbackAwaiting parks a delivered callback effect awaiting the frozen
// responder's reply. A Task-watch handoff makes the callback runnable
// immediately so the resolved responder enters the Run frontier right after the
// ordinary Task transaction commits; other callbacks wait out the parked delay.
func (dispatcher *EffectDispatcher) resolveCallbackAwaiting(
	ctx context.Context,
	effect workflowrunstore.Effect,
	request EffectRequest,
	result MessageResult,
	receipt EffectResult,
	completedAt time.Time,
) error {
	payload, err := EncodeEffectResult(receipt)
	if err != nil {
		return err
	}
	delivered := completedAt
	deadline := completedAt.Add(parkedRunDelay)
	runnableAt := deadline
	if request.Message.TaskWatchHandoff != nil {
		// Freeze the resolved responder into the Run frontier immediately
		// after the ordinary Task transaction commits. The engine then parks
		// this callback until the Agent reply or a delivery failure.
		runnableAt = completedAt
	}
	_, _, err = dispatcher.repository.ResolveEffect(ctx, workflowrunstore.ResolveEffectParams{
		WorkspaceID: effect.WorkspaceID, RunID: effect.RunID, EffectID: effect.ID, LeaseToken: effect.LeaseToken,
		Status: workflowrunstore.EffectStatusAwaiting, CallbackResponderAgentID: receipt.CallbackResponderAgentID,
		ResultJSON: payload, DeliveredAt: &delivered,
		RunnableAt: runnableAt, Events: []workflowrunstore.EventDraft{effectEvent("message_callback_awaiting", effect, request, map[string]any{
			"messageId": result.MessageID, "wakeAgentIds": result.WakeAgentIDs,
			"callbackResponderAgentId": receipt.CallbackResponderAgentID, "deliveryScope": result.DeliveryScope,
		})}, Now: completedAt,
	})
	return err
}

func (dispatcher *EffectDispatcher) validateCallbackResponderReceipt(
	ctx context.Context,
	effect workflowrunstore.Effect,
	request EffectRequest,
	receipt EffectResult,
	frozenResponderAgentID string,
) error {
	// The frozen responder is authoritative for fresh writes; the persisted
	// effect ID is authoritative for bound receipt recovery.
	authoritativeResponder := strings.TrimSpace(frozenResponderAgentID)
	if authoritativeResponder == "" {
		authoritativeResponder = strings.TrimSpace(effect.CallbackResponderAgentID)
	}
	receiptResponder := strings.TrimSpace(receipt.CallbackResponderAgentID)
	if request.Intent.Callback && authoritativeResponder != "" && receiptResponder != "" && receiptResponder != authoritativeResponder {
		return dispatcher.resolveCallbackResponderChanged(ctx, effect, request, receipt, authoritativeResponder)
	}
	return nil
}

// gateCallbackDelivery resolves the resolved callback responder through the
// trusted chain only when the effect has not yet frozen one; once frozen, the
// persisted resolved Agent ID is the sole authority and is never re-resolved
// from a mutable AgentRef. It atomically acquires the (workspace, Agent)
// execution lease or enqueues the effect. It returns gated=false when the
// effect is durably queued_for_agent (the caller must NOT deliver), and the
// frozen responder ID when delivery may proceed so the writer receipt can be
// verified against it.
func (dispatcher *EffectDispatcher) materializeMessageContent(ctx context.Context, message *MessageRequest) error {
	if message == nil || message.ContentRef == nil {
		return nil
	}
	resolver, ok := dispatcher.repository.(ArtifactContentResolver)
	if !ok {
		return fmt.Errorf("workflow artifact content resolver is not configured")
	}
	payload, err := resolver.ResolveArtifactContent(ctx, *message.ContentRef)
	if err != nil {
		return err
	}
	var content string
	if err := json.Unmarshal(payload, &content); err != nil || strings.TrimSpace(content) == "" {
		return fmt.Errorf("workflow artifact message content is invalid")
	}
	message.Content = content
	message.ContentRef = nil
	return message.Validate()
}

func (dispatcher *EffectDispatcher) gateCallbackDelivery(ctx context.Context, effect workflowrunstore.Effect, request EffectRequest) (bool, string, error) {
	if request.Message == nil {
		return false, "", fmt.Errorf("workflow Run callback requires a message request")
	}
	if len(request.Message.ResponderCandidates) > 0 {
		return dispatcher.gateSelectOneCallbackDelivery(ctx, effect, request)
	}
	responderAgentID := strings.TrimSpace(effect.CallbackResponderAgentID)
	if responderAgentID == "" {
		if dispatcher.responders == nil {
			return false, "", fmt.Errorf("workflow Run callback responder resolver is not configured")
		}
		resolved, err := dispatcher.responders.ResolveCallbackResponder(ctx, *request.Message)
		if err != nil {
			return false, "", err
		}
		responderAgentID = strings.TrimSpace(resolved)
		if responderAgentID == "" {
			return false, "", fmt.Errorf("workflow Run callback responder resolved to an empty Agent")
		}
	}
	result, err := dispatcher.repository.ClaimCallbackEffectCapacity(ctx, workflowrunstore.ClaimCallbackEffectCapacityParams{
		WorkspaceID: effect.WorkspaceID, RunID: effect.RunID, ActivationID: effect.ActivationID, EffectID: effect.ID,
		WorkerID: dispatcher.workerID, Now: dispatcher.clockNow(), LeaseTTL: dispatcher.leaseTTL,
		LeaseToken: effect.LeaseToken, ResponderAgentID: responderAgentID, NodeID: request.Intent.NodeID,
	})
	if err != nil {
		return false, "", err
	}
	if result.Queued {
		return false, "", nil
	}
	if !result.Claimed {
		return false, "", fmt.Errorf("workflow Run callback capacity decision is indeterminate")
	}
	return true, responderAgentID, nil
}

// gateSelectOneCallbackDelivery resolves the resolved callback responder for a
// responder.selectOne callback from the frozen ordered candidate pool. It
// atomically selects the highest-priority candidate with free capacity or
// durably parks the effect as waiting_for_responder. A parked (Waiting) effect
// must NOT deliver: the message transaction is never started, so an all-busy
// pool produces no writer side effects and no Task watch. Once a candidate is
// selected its ID is frozen on the effect and is the sole authority — it is
// never re-resolved from a mutable AgentRef or re-selected after a freeze.
func (dispatcher *EffectDispatcher) gateSelectOneCallbackDelivery(ctx context.Context, effect workflowrunstore.Effect, request EffectRequest) (bool, string, error) {
	if request.Message == nil {
		return false, "", fmt.Errorf("workflow Run callback requires a message request")
	}
	result, err := dispatcher.repository.SelectCallbackResponderCapacity(ctx, workflowrunstore.SelectCallbackResponderCapacityParams{
		WorkspaceID: effect.WorkspaceID, RunID: effect.RunID, ActivationID: effect.ActivationID, EffectID: effect.ID,
		WorkerID: dispatcher.workerID, Now: dispatcher.clockNow(), LeaseTTL: dispatcher.leaseTTL,
		LeaseToken: effect.LeaseToken, NodeID: request.Intent.NodeID,
		CandidateResponderAgentIDs: append([]string(nil), request.Message.ResponderCandidates...),
		SelectOneStrategy:          request.Message.SelectOneStrategy,
		Randomizer:                 dispatcher.randomizer,
	})
	if err != nil {
		return false, "", err
	}
	if result.Waiting {
		// Every candidate is busy; the effect is durably parked
		// waiting_for_responder. Do NOT deliver and do NOT start a message
		// transaction (no writer side effects while the pool is fully busy).
		return false, "", nil
	}
	if !result.Claimed {
		return false, "", fmt.Errorf("workflow Run callback capacity decision is indeterminate")
	}
	selected := strings.TrimSpace(result.SelectedResponderAgentID)
	if selected == "" {
		return false, "", fmt.Errorf("workflow Run callback candidate selection resolved to an empty Agent")
	}
	return true, selected, nil
}

// resolveCallbackResponderChanged fences a callback whose ordinary writer
// receipt named a different Agent than the pre-frozen responder. The message
// is already visible and bound, so the terminal failed receipt preserves the
// bound message id and the frozen responder for audit rather than retrying or
// silently retargeting.
func (dispatcher *EffectDispatcher) resolveCallbackResponderChanged(ctx context.Context, effect workflowrunstore.Effect, request EffectRequest, receipt EffectResult, frozenResponderAgentID string) error {
	now := dispatcher.clockNow()
	failed := EffectResult{
		Port: "failed", Output: runkernel.Values{},
		FailureCode: "callback.responder_changed", FailureMessage: "writer receipt responder does not match the frozen callback responder",
		FailureProblem: runkernel.FailureProblem("callback.responder_changed"),
		FailureHint:    runkernel.FailureHint("callback.responder_changed"),
		FailureStage:   "callback", FailurePath: request.Intent.NodeID,
		MessageID: receipt.MessageID, CallbackResponderAgentID: frozenResponderAgentID,
	}
	payload, err := EncodeEffectResult(failed)
	if err != nil {
		return err
	}
	completed := now
	_, _, err = dispatcher.repository.ResolveEffect(ctx, workflowrunstore.ResolveEffectParams{
		WorkspaceID: effect.WorkspaceID, RunID: effect.RunID, EffectID: effect.ID, LeaseToken: effect.LeaseToken,
		Status: workflowrunstore.EffectStatusFailed, ResultJSON: payload, ErrorCode: "callback.responder_changed",
		ErrorJSON: failureJSON(failed), CompletedAt: &completed, RunnableAt: now,
		Events: []workflowrunstore.EventDraft{failureEffectEvent("effect_failed", effect, request, failed, nil)}, Now: now,
	})
	return err
}

func validatedMessageReceipt(request EffectRequest, result MessageResult, boundMessageID string) (EffectResult, error) {
	if err := result.Validate(); err != nil {
		return EffectResult{}, err
	}
	if boundMessageID != "" && result.MessageID != boundMessageID {
		return EffectResult{}, fmt.Errorf("workflow Run recovered message receipt does not match bound message")
	}
	callbackResponderAgentID, err := ResolveCallbackResponderAuthority(
		request.Intent.Callback,
		result.CallbackResponderAgentID,
		result.WakeAgentIDs,
	)
	if err != nil {
		return EffectResult{}, err
	}
	if request.Message.Scope != nil && *result.DeliveryScope != *request.Message.Scope {
		return EffectResult{}, fmt.Errorf("workflow Run message result deliveryScope does not match requested scope")
	}
	if err := validateMessageThreadReceipt(request.Intent.StartThread, result.ThreadScope); err != nil {
		return EffectResult{}, err
	}
	return EffectResult{
		MessageID:                result.MessageID,
		WakeAgentIDs:             result.WakeAgentIDs,
		CallbackResponderAgentID: callbackResponderAgentID,
		DeliveryScope:            result.DeliveryScope,
		ThreadScope:              result.ThreadScope,
		TeamMemberContext:        cloneCallbackTeamMemberContext(result.TeamMemberContext),
	}, nil
}

func validateMessageThreadReceipt(startThread bool, threadScope *domain.MessageScopeValue) error {
	if startThread && threadScope == nil {
		return fmt.Errorf("workflow Run message result threadScope is required")
	}
	if !startThread && threadScope != nil {
		return fmt.Errorf("workflow Run message result cannot include threadScope")
	}
	return nil
}

func (dispatcher *EffectDispatcher) dispatchApproval(ctx context.Context, effect workflowrunstore.Effect, request EffectRequest) error {
	if dispatcher.approvals == nil || dispatcher.assignees == nil {
		return dispatcher.resolveFailure(ctx, effect, request, "approval.writer_unavailable", fmt.Errorf("Workflow App approval dependencies are not configured"), false)
	}
	run, err := dispatcher.repository.LoadRun(ctx, effect.WorkspaceID, effect.RunID)
	if err != nil {
		return dispatcher.resolveFailure(ctx, effect, request, "approval.run_unavailable", err, true)
	}
	assigneeMemberID := strings.TrimSpace(effect.ApprovalAssigneeMemberID)
	if assigneeMemberID == "" {
		assigneeMemberID, err = dispatcher.assignees.ResolveWorkflowApprovalAssignee(ctx, effect.WorkspaceID, request.Intent.Assignee)
		if err != nil {
			return dispatcher.resolveFailure(ctx, effect, request, "approval.assignee_unavailable", err, true)
		}
		// Freeze the exact Human while the dispatch lease is still live and
		// before the Workflow App validates/creates its DM notification. A
		// reclaimed attempt reads this value back instead of re-resolving a
		// mutable alias that may now name somebody else.
		if err := dispatcher.repository.FreezeApprovalAssignee(
			ctx, effect.WorkspaceID, effect.RunID, effect.ID, effect.LeaseToken, assigneeMemberID,
		); err != nil {
			return err
		}
	}
	receipt, err := dispatcher.approvals.WriteApproval(ctx, ApprovalRequest{
		WorkspaceID: effect.WorkspaceID, PlanID: run.PlanID, PlanVersionID: run.PlanVersionID,
		RunID: run.ID, ActivationID: request.Intent.ActivationID, EffectID: effect.ID, NodeID: request.Intent.NodeID,
		AssigneeMemberID: assigneeMemberID, Summary: request.Intent.Summary,
		FeedbackScope:    cloneMessageScopeValue(request.Intent.FeedbackScope),
		CausedByMemberID: run.Identity.AuthorMemberID,
	})
	if err != nil {
		return dispatcher.resolveFailure(ctx, effect, request, "approval.delivery_failed", err, true)
	}
	completedAt := dispatcher.clockNow()
	result, err := EncodeEffectResult(EffectResult{
		NotificationID: receipt.NotificationID, NotificationRevision: receipt.NotificationRevision,
		MessageID: receipt.MessageID, AssigneeMemberID: assigneeMemberID,
	})
	if err != nil {
		return err
	}
	delivered := completedAt
	runnableAt := completedAt.Add(parkedRunDelay)
	_, _, err = dispatcher.repository.ResolveEffect(ctx, workflowrunstore.ResolveEffectParams{
		WorkspaceID: effect.WorkspaceID, RunID: effect.RunID, EffectID: effect.ID, LeaseToken: effect.LeaseToken,
		Status: workflowrunstore.EffectStatusAwaiting, ResultJSON: result, DeliveredAt: &delivered,
		RunnableAt: runnableAt,
		Events:     []workflowrunstore.EventDraft{effectEvent("approval_awaiting", effect, request, map[string]any{"notificationId": receipt.NotificationID, "assigneeMemberId": assigneeMemberID})}, Now: completedAt,
	})
	return err
}

func (dispatcher *EffectDispatcher) resolveFailure(ctx context.Context, effect workflowrunstore.Effect, request EffectRequest, code string, cause error, retryable bool) error {
	now := dispatcher.clockNow()
	if effect.Kind == workflowrunstore.EffectKindMessage && strings.TrimSpace(effect.MessageID) != "" {
		// The message is already visible, so maxAttempts and the authored
		// delivery deadline no longer describe whether delivery may be retried;
		// they only bounded the original write. Keep replaying with the stable
		// clientMessageId until the native writer reconstructs the committed
		// receipt. This also covers a temporarily unavailable writer or an
		// invalid/incomplete replay response without ever manufacturing a
		// message-less terminal result.
		backoff := time.Duration(request.Intent.Retry.BackoffNanos)
		if backoff < time.Second {
			backoff = time.Second
		}
		retryAt := now.Add(backoff).UTC().Truncate(time.Microsecond)
		errorJSON := failureJSON(EffectResult{
			FailureCode: code, FailureProblem: runkernel.FailureProblem(code), FailureHint: runkernel.FailureHint(code), FailureRetryable: true,
			FailureStage: "dispatch", FailurePath: request.Intent.NodeID,
		})
		_, _, err := dispatcher.repository.ResolveEffect(ctx, workflowrunstore.ResolveEffectParams{
			WorkspaceID: effect.WorkspaceID, RunID: effect.RunID, EffectID: effect.ID, LeaseToken: effect.LeaseToken,
			Status: workflowrunstore.EffectStatusPending, ErrorCode: code, ErrorJSON: errorJSON, RetryAt: &retryAt,
			RunnableAt: retryAt, Events: []workflowrunstore.EventDraft{failureEffectEvent("effect_retry_scheduled", effect, request, EffectResult{
				FailureCode: code, FailureProblem: runkernel.FailureProblem(code), FailureHint: runkernel.FailureHint(code), FailureRetryable: true,
				FailureStage: "dispatch", FailurePath: request.Intent.NodeID,
			}, map[string]any{"retryAt": retryAt, "messageId": effect.MessageID, "receiptRecovery": true})}, Now: now,
		})
		return err
	}
	deadline := effect.DeadlineAt
	if deadline == nil {
		deadline = request.Intent.Deadline
	}
	if retryable && int64(request.Intent.Retry.MaxAttempts) > effect.AttemptCount {
		retryAt := now.Add(time.Duration(request.Intent.Retry.BackoffNanos)).UTC().Truncate(time.Microsecond)
		if deadline == nil || retryAt.Before(*deadline) {
			errorJSON := failureJSON(EffectResult{
				FailureCode: code, FailureProblem: runkernel.FailureProblem(code), FailureHint: runkernel.FailureHint(code), FailureRetryable: true,
				FailureStage: "dispatch", FailurePath: request.Intent.NodeID,
			})
			runnable := retryAt
			if deadline != nil {
				runnable = deadline.UTC().Truncate(time.Microsecond)
			}
			_, _, err := dispatcher.repository.ResolveEffect(ctx, workflowrunstore.ResolveEffectParams{
				WorkspaceID: effect.WorkspaceID, RunID: effect.RunID, EffectID: effect.ID, LeaseToken: effect.LeaseToken,
				Status: workflowrunstore.EffectStatusPending, ErrorCode: code, ErrorJSON: errorJSON, RetryAt: &retryAt,
				RunnableAt: runnable, Events: []workflowrunstore.EventDraft{failureEffectEvent("effect_retry_scheduled", effect, request, EffectResult{
					FailureCode: code, FailureProblem: runkernel.FailureProblem(code), FailureHint: runkernel.FailureHint(code), FailureRetryable: true,
					FailureStage: "dispatch", FailurePath: request.Intent.NodeID,
				}, map[string]any{"retryAt": retryAt})}, Now: now,
			})
			return err
		}
	}
	// The terminal dispatch failure persists only the public-safe
	// contract: the stable code plus a bounded problem and hint derived from it.
	// The arbitrary cause (which may embed SQL, credentials, stack traces, or
	// tenant identifiers) stays in internal observability and never crosses the
	// persisted error boundary via FailureMessage or ErrorJSON.
	result := EffectResult{
		Port: "failed", Output: runkernel.Values{}, FailureCode: code,
		FailureMessage: runkernel.FailureHint(code), FailureProblem: runkernel.FailureProblem(code),
		FailureHint: runkernel.FailureHint(code), FailureRetryable: retryable,
		FailureStage: "dispatch", FailurePath: request.Intent.NodeID,
	}
	return dispatcher.resolvePort(ctx, effect, result, workflowrunstore.EffectStatusFailed,
		failureEffectEvent("effect_failed", effect, request, result, nil), now, now)
}

// dispatchFailureHint returns a stable, bounded public-safe hint for an effect
// dispatch failure code. The arbitrary cause stays in logs only and never
// crosses the persisted error boundary.

func (dispatcher *EffectDispatcher) resolvePort(ctx context.Context, effect workflowrunstore.Effect, result EffectResult, status string, event workflowrunstore.EventDraft, runnableAt, now time.Time) error {
	now = now.UTC().Truncate(time.Microsecond)
	runnableAt = runnableAt.UTC().Truncate(time.Microsecond)
	result.CompletedAt = &now
	payload, err := EncodeEffectResult(result)
	if err != nil {
		return err
	}
	completed := now
	_, _, err = dispatcher.repository.ResolveEffect(ctx, workflowrunstore.ResolveEffectParams{
		WorkspaceID: effect.WorkspaceID, RunID: effect.RunID, EffectID: effect.ID, LeaseToken: effect.LeaseToken,
		Status: status, ResultJSON: payload, ErrorCode: result.FailureCode,
		ErrorJSON: failureJSON(result), CompletedAt: &completed, RunnableAt: runnableAt,
		Events: []workflowrunstore.EventDraft{event}, Now: now,
	})
	return err
}

func effectDeadlineExpired(effect workflowrunstore.Effect, request EffectRequest, now time.Time) bool {
	deadline := request.Intent.Deadline
	if effect.DeadlineAt != nil {
		deadline = effect.DeadlineAt
	}
	return deadline != nil && !deadline.After(now)
}

func (dispatcher *EffectDispatcher) resolveTimedOut(ctx context.Context, effect workflowrunstore.Effect, request EffectRequest, now time.Time) error {
	// resolveTimedOut fires only after the authored deadline has already
	// expired (effectDeadlineExpired), so the Run is terminal for this effect:
	// the outcome is a definitive timeout, not a transient failure that a retry
	// could resolve. Mark it non-retryable while transient pre-deadline dispatch
	// failures (resolveFailure retry path) keep retryable=true.
	result := EffectResult{
		Port: "timed_out", Output: runkernel.Values{}, FailureCode: "effect.timed_out",
		FailureProblem: runkernel.FailureProblem("effect.timed_out"),
		FailureHint:    runkernel.FailureHint("effect.timed_out"),
		FailureStage:   "dispatch", FailurePath: request.Intent.NodeID,
		FailureRetryable: false,
	}
	return dispatcher.resolvePort(ctx, effect, result, workflowrunstore.EffectStatusCompleted,
		failureEffectEvent("effect_timed_out", effect, request, result, nil), now, now)
}

func effectEvent(kind string, effect workflowrunstore.Effect, request EffectRequest, extra map[string]any) workflowrunstore.EventDraft {
	payload := map[string]any{
		"effectId": effect.ID, "activationId": request.Intent.ActivationID,
		"nodeId": request.Intent.NodeID, "kind": request.Intent.Kind,
	}
	for key, value := range extra {
		payload[key] = value
	}
	raw, _ := json.Marshal(payload)
	return workflowrunstore.EventDraft{Type: kind, PayloadJSON: raw}
}

// failureEffectEvent builds an effect failure/retry event whose payload carries
// the public safe failure contract (code/stage/path/problem/hint/retryable/
// workItem/diagnostics) so the persisted event and its Run log Details
// projection preserve every field the Agent/CLI needs. Diagnostics are routed
// through the single shared diagnostic budget so the event carries the exact
// same deterministic cuts and truncation metadata (omittedCount/truncatedFields/
// serializedBytes) as the persisted error_json; prior truncation from a replayed
// EffectResult is carried forward rather than recomputed away. extra may still
// carry operationally safe event-specific keys (retryAt, messageId,
// receiptRecovery); operational identifiers never enter the payload.
func failureEffectEvent(kind string, effect workflowrunstore.Effect, request EffectRequest, result EffectResult, extra map[string]any) workflowrunstore.EventDraft {
	payload := map[string]any{
		"effectId": effect.ID, "activationId": request.Intent.ActivationID,
		"nodeId": request.Intent.NodeID, "kind": request.Intent.Kind,
		"code": result.FailureCode,
	}
	if result.FailureStage != "" {
		payload["stage"] = result.FailureStage
	}
	if result.FailurePath != "" {
		payload["path"] = result.FailurePath
	}
	payload["problem"] = firstNonEmpty(result.FailureProblem, runkernel.FailureProblem(result.FailureCode))
	if result.FailureHint != "" {
		payload["hint"] = result.FailureHint
	} else if hint := runkernel.FailureHint(result.FailureCode); hint != "" {
		payload["hint"] = hint
	}
	if result.FailureRetryable {
		payload["retryable"] = true
	}
	// Always route through the shared carried budget so a replayed/all-dropped
	// failure still emits its prior truncation metadata even when the diagnostics
	// list is empty; an empty list never fabricates diagnostics but preserves the
	// earlier omittedCount/truncatedFields/serializedBytes in the event payload.
	bounded, truncation := boundPersistedDiagnosticsCarried(result.FailureDiagnostics, result.FailureTruncation)
	if len(bounded) > 0 {
		payload["diagnostics"] = bounded
	}
	if truncation != nil {
		payload["truncation"] = truncation
	}
	if result.FailureWorkItem != nil {
		payload["workItem"] = result.FailureWorkItem
	}
	for key, value := range extra {
		if key == "code" || key == "stage" || key == "path" || key == "problem" || key == "hint" || key == "retryable" || key == "diagnostics" || key == "workItem" || key == "truncation" {
			continue
		}
		payload[key] = value
	}
	raw, _ := json.Marshal(payload)
	return workflowrunstore.EventDraft{Type: kind, PayloadJSON: raw}
}

func failureJSON(result EffectResult) json.RawMessage {
	if result.FailureCode == "" && result.FailureMessage == "" {
		return nil
	}
	persisted := newPersistedFailure(
		result.FailureCode,
		result.FailureStage,
		result.FailurePath,
		firstNonEmpty(result.FailureProblem, runkernel.FailureProblem(result.FailureCode)),
		firstNonEmpty(result.FailureHint, runkernel.FailureHint(result.FailureCode)),
		result.FailureRetryable,
		result.FailureDiagnostics,
		result.FailureWorkItem,
		result.FailureTruncation,
	)
	payload, _ := encodePersistedFailure(persisted)
	return payload
}