package orchestrator

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

	"github.com/tidbcloud/AgentTeam/internal/core/domain"
	"github.com/tidbcloud/AgentTeam/internal/coredb"
	"github.com/tidbcloud/AgentTeam/internal/store/storeerr"
	workflowrunstore "github.com/tidbcloud/AgentTeam/internal/store/workflow/run"
	workflowpayload "github.com/tidbcloud/AgentTeam/internal/workflow/payload"
	"github.com/tidbcloud/AgentTeam/internal/workflow/plan"
	"github.com/tidbcloud/AgentTeam/internal/workflow/publicerr"
	runkernel "github.com/tidbcloud/AgentTeam/internal/workflow/run"
	"github.com/tidbcloud/AgentTeam/internal/workflow/runartifact"
)

const (
	callbackCompletedPort = "completed"
	callbackFailedPort    = "failed"
)

// CallbackRepository is the smallest persistence boundary needed by the
// authenticated Agent callback surface. ResolveAwaitingEffect owns the
// first-result-wins transaction; all identity carried by CallbackCommand is
// independently rechecked against the frozen effect receipt before it is
// allowed to reach that transaction.
type CallbackRepository interface {
	LoadRun(context.Context, string, string) (workflowrunstore.Run, error)
	LoadActivation(context.Context, string, string, string) (workflowrunstore.Activation, error)
	LoadEffect(context.Context, string, string, string) (workflowrunstore.Effect, error)
	ListAwaitingCallbackEffects(context.Context, workflowrunstore.ListAwaitingCallbackEffectsParams) ([]workflowrunstore.Effect, error)
	ResolveAwaitingEffect(context.Context, workflowrunstore.ResolveAwaitingEffectParams) (workflowrunstore.ResolveAwaitingEffectResult, error)
	// PauseAwaitingCallbackRecovery implements the non-terminal `pause` action:
	// the awaiting effect stays awaiting but automatic recovery stops durably
	// and permanently (no resume trigger; only complete/fail/cancel/terminal
	// lifecycle cleans it).
	PauseAwaitingCallbackRecovery(context.Context, string, string, string, time.Time) (bool, error)
	PauseAwaitingCallbackRecoveryWithIdempotency(context.Context, string, string, string, time.Time, string) (workflowrunstore.CallbackRecoveryActionResult, error)
	// ParkAwaitingCallbackRecovery implements the non-terminal `park` action: it
	// arms the durable park FSM so automatic recovery stops while the responder
	// is idle and resumes on a fresh schedule after the first authoritative
	// working observation. A paused callback cannot be parked.
	ParkAwaitingCallbackRecovery(context.Context, string, string, string, time.Time) (bool, error)
	ParkAwaitingCallbackRecoveryWithIdempotency(context.Context, string, string, string, time.Time, string) (workflowrunstore.CallbackRecoveryActionResult, error)
}

// CallbackAuthorityRepository is implemented by the production Run store. It
// is intentionally additive so legacy in-memory callback adapters remain
// usable in unit tests with an empty MemberID, while authenticated production
// requests fail closed if the durable authority boundary is absent.
type CallbackAuthorityRepository interface {
	AuthorizeCallbackAgent(context.Context, workflowrunstore.AgentPrincipal, string, string, string) error
}

// CallbackAuthorityTxRepository is the transaction-aware callback boundary used
// by task-watch mutations and callback completion precommit guards.
type CallbackAuthorityTxRepository interface {
	AuthorizeCallbackAgentTx(context.Context, coredb.Tx, workflowrunstore.AgentPrincipal, string, string, string) error
}

// CallbackRecoveryMutationRepository adds principal-bound pause/park writes to
// the compatibility recovery repository. Production implements these methods;
// old adapters are only accepted for commands without a MemberID.
type CallbackRecoveryMutationRepository interface {
	PauseAwaitingCallbackRecoveryForAgent(context.Context, workflowrunstore.CallbackRecoveryMutationParams) (workflowrunstore.CallbackRecoveryActionResult, error)
	ParkAwaitingCallbackRecoveryForAgent(context.Context, workflowrunstore.CallbackRecoveryMutationParams) (workflowrunstore.CallbackRecoveryActionResult, error)
}

// CallbackScopeAuthorizer is the narrow callback boundary for normalizing
// and authorizing scope-valued Agent output. Implementations must authorize the
// exact authenticated Agent and the frozen Human invocation owner against the
// same active scope before returning its normalized value.
// CallbackArtifactReader is an additive capability implemented by the
// production Workflow Run store. References are checked against the exact
// workspace/run/producer binding on every page request.
type CallbackArtifactReader interface {
	ReadArtifactPage(context.Context, runartifact.Ref, string, []string, string, int) (workflowrunstore.ArtifactPage, error)
}

type CallbackScopeAuthorizer interface {
	AuthorizeWorkflowRunCallbackScope(
		context.Context,
		string,
		string,
		string,
		string,
		domain.MessageScopeValue,
	) (domain.MessageScopeValue, error)
	AuthorizeWorkflowRunCallbackScopeTx(
		context.Context,
		coredb.Tx,
		string,
		string,
		string,
		string,
		domain.MessageScopeValue,
	) (domain.MessageScopeValue, error)
	AuthorizeWorkflowRunCallbackRelatedScopes(
		context.Context,
		string,
		string,
		string,
		string,
		[]domain.MessageScopeValue,
	) ([]domain.MessageScopeValue, error)
	AuthorizeWorkflowRunCallbackRelatedScopesTx(
		context.Context,
		coredb.Tx,
		string,
		string,
		string,
		string,
		[]domain.MessageScopeValue,
	) ([]domain.MessageScopeValue, error)
}

// CallbackTaskWatchMutator promotes the authenticated callback Agent to a
// Task Watcher after an agent-resolved Task workScope has been authorized.
// It runs inside ResolveAwaitingEffect's first-result-wins transaction, so the
// Watch mutation and callback completion commit or roll back together.
type CallbackTaskWatchMutator interface {
	ActivateWorkflowRunCallbackTaskWatch(
		context.Context,
		coredb.Tx,
		string,
		string,
		domain.MessageScopeValue,
	) error
}

type CallbackServiceOptions struct {
	Now              func() time.Time
	ScopeAuthorizer  CallbackScopeAuthorizer
	TaskWatchMutator CallbackTaskWatchMutator
}

// CallbackService authorizes and durably completes send_message.callback
// effects. In particular, it never resolves the current Agent from mutable
// Plan/Binding data: only CallbackResponderAgentID, frozen separately in the
// durable message receipt, grants callback authority. Ordinary routing may
// wake additional Agents without authorizing their callbacks.
type CallbackService struct {
	repository       CallbackRepository
	scopeAuthorizer  CallbackScopeAuthorizer
	taskWatchMutator CallbackTaskWatchMutator
	now              func() time.Time
}

func NewCallbackService(repository CallbackRepository, options CallbackServiceOptions) (*CallbackService, error) {
	if repository == nil {
		return nil, fmt.Errorf("workflow Run callback repository is required")
	}
	if options.Now == nil {
		options.Now = time.Now
	}
	return &CallbackService{
		repository:       repository,
		scopeAuthorizer:  options.ScopeAuthorizer,
		taskWatchMutator: options.TaskWatchMutator,
		now:              options.Now,
	}, nil
}

func (service *CallbackService) authorizeCallbackCommand(ctx context.Context, command CallbackCommand) error {
	if strings.TrimSpace(command.MemberID) == "" {
		// Compatibility adapters may omit the authenticated member. The public
		// Agent transport always supplies it and crosses the lifecycle boundary.
		return nil
	}
	authorizer, ok := service.repository.(CallbackAuthorityRepository)
	if !ok {
		return ErrCallbackUnauthorized
	}
	if err := authorizer.AuthorizeCallbackAgent(ctx, workflowrunstore.AgentPrincipal{
		WorkspaceID: command.WorkspaceID, AgentID: command.AgentID, MemberID: command.MemberID,
	}, command.WorkspaceID, command.RunID, command.CallbackID); err != nil {
		if errors.Is(err, storeerr.ErrNotFound) {
			// The durable authority predicate intentionally collapses a foreign
			// responder into not-found. Once the caller has supplied the exact
			// workspace/run/effect selector, however, preserving the distinction
			// is useful for the authenticated Agent transport: an existing effect
			// whose frozen responder differs is a forbidden callback, while an
			// absent selector remains non-enumerating not-found.
			if effect, loadErr := service.repository.LoadEffect(ctx, command.WorkspaceID, command.RunID, command.CallbackID); loadErr == nil &&
				effect.CallbackResponderAgentID != command.AgentID {
				return ErrCallbackUnauthorized
			}
			return ErrCallbackNotFound
		}
		return err
	}
	return nil
}

func (service *CallbackService) ListCallbacks(
	ctx context.Context,
	command CallbackListCommand,
) (CallbackListResult, error) {
	if service == nil || service.repository == nil || service.now == nil {
		return CallbackListResult{}, fmt.Errorf("workflow Run callback service is not configured")
	}
	command.WorkspaceID = strings.TrimSpace(command.WorkspaceID)
	command.AgentID = strings.TrimSpace(command.AgentID)
	command.MemberID = strings.TrimSpace(command.MemberID)
	command.Cursor = strings.TrimSpace(command.Cursor)
	if command.WorkspaceID == "" || command.AgentID == "" || command.Limit < 1 || command.Limit > 100 {
		return CallbackListResult{}, ErrCallbackNotFound
	}
	effects, err := service.repository.ListAwaitingCallbackEffects(ctx, workflowrunstore.ListAwaitingCallbackEffectsParams{
		WorkspaceID: command.WorkspaceID,
		AgentID:     command.AgentID,
		MemberID:    command.MemberID,
		AfterID:     command.Cursor,
		Limit:       command.Limit + 1,
	})
	if err != nil {
		return CallbackListResult{}, mapCallbackReadError(err)
	}
	nextCursor := ""
	if len(effects) > command.Limit {
		effects = effects[:command.Limit]
		nextCursor = effects[len(effects)-1].ID
	}
	now := service.now().UTC().Truncate(time.Microsecond)
	results := make([]CallbackResult, 0, len(effects))
	for _, effect := range effects {
		result, include, err := service.listCallbackResult(ctx, command, effect, now)
		if err != nil {
			return CallbackListResult{}, err
		}
		if include {
			results = append(results, result)
		}
	}
	return CallbackListResult{Steps: results, NextCursor: nextCursor}, nil
}

func (service *CallbackService) listCallbackResult(
	ctx context.Context,
	command CallbackListCommand,
	effect workflowrunstore.Effect,
	now time.Time,
) (CallbackResult, bool, error) {
	// This equality is redundant with the indexed repository predicate, but
	// keeps the service boundary fail closed for alternate implementations.
	if effect.CallbackResponderAgentID != command.AgentID {
		return CallbackResult{}, false, nil
	}
	if command.MemberID != "" {
		if err := service.authorizeCallbackCommand(ctx, CallbackCommand{
			WorkspaceID: command.WorkspaceID, RunID: effect.RunID, CallbackID: effect.ID,
			AgentID: command.AgentID, MemberID: command.MemberID,
		}); err != nil {
			if errors.Is(err, ErrCallbackNotFound) || errors.Is(err, ErrCallbackUnauthorized) {
				return CallbackResult{}, false, nil
			}
			return CallbackResult{}, false, err
		}
	}
	run, err := service.repository.LoadRun(ctx, command.WorkspaceID, effect.RunID)
	if err != nil {
		return CallbackResult{}, false, mapCallbackReadError(err)
	}
	activation, err := service.repository.LoadActivation(ctx, command.WorkspaceID, effect.RunID, effect.ActivationID)
	if err != nil {
		return CallbackResult{}, false, mapCallbackReadError(err)
	}
	snapshot, callbackErr := callbackSnapshotFromRecords(run, activation, effect, now)
	if callbackErr == nil && snapshot.receipt.CallbackResponderAgentID != command.AgentID {
		callbackErr = ErrCallbackUnauthorized
	}
	if callbackErr == nil {
		result, resultErr := callbackResultFromSnapshot(snapshot, false)
		return result, resultErr == nil, resultErr
	}
	if errors.Is(callbackErr, ErrCallbackUnauthorized) ||
		errors.Is(callbackErr, ErrCallbackNotFound) ||
		errors.Is(callbackErr, ErrCallbackNotReady) ||
		errors.Is(callbackErr, ErrCallbackNotWaiting) ||
		errors.Is(callbackErr, ErrRunTerminal) {
		return CallbackResult{}, false, nil
	}
	return CallbackResult{}, false, callbackErr
}

type callbackSnapshot struct {
	run        workflowrunstore.Run
	activation workflowrunstore.Activation
	effect     workflowrunstore.Effect
	request    EffectRequest
	receipt    EffectResult
}

// ReadCallbackSection returns one lossless, bounded section of an authorized
// callback receipt. It reuses the same frozen responder/Run/Step authorization
// as get/complete and never exposes a section from a foreign workspace.
func (service *CallbackService) ReadCallbackSection(ctx context.Context, command CallbackCommand) (CallbackPayloadRead, error) {
	if service == nil || service.repository == nil || service.now == nil {
		return CallbackPayloadRead{}, fmt.Errorf("workflow Run callback service is not configured")
	}
	command = normalizeCallbackCommand(command)
	if err := validateCallbackCommand(command); err != nil {
		return CallbackPayloadRead{}, err
	}
	if command.Action != CallbackActionGetCurrent || command.Section == "" {
		return CallbackPayloadRead{}, storeerr.BadRequest("callback payload section is required")
	}
	if command.Section == "compact" {
		return CallbackPayloadRead{}, storeerr.BadRequest("callback payload section is invalid")
	}
	if err := workflowpayload.ValidateReadSelector(command.Section, command.Path, command.Cursor); err != nil {
		return CallbackPayloadRead{}, storeerr.BadRequest(err.Error())
	}
	if command.Limit == 0 {
		command.Limit = workflowpayload.MaxChunkBytes
	}
	if command.Limit < 1 || command.Limit > workflowpayload.MaxChunkBytes {
		return CallbackPayloadRead{}, storeerr.BadRequest("callback payload limit is invalid")
	}
	now := service.now().UTC().Truncate(time.Microsecond)
	snapshot, err := service.loadSnapshot(ctx, command, now)
	if err != nil {
		return CallbackPayloadRead{}, err
	}
	if snapshot.receipt.CallbackResponderAgentID != command.AgentID {
		return CallbackPayloadRead{}, ErrCallbackUnauthorized
	}
	if snapshot.run.Terminal() || snapshot.effect.Status != workflowrunstore.EffectStatusAwaiting ||
		snapshot.activation.Status != workflowrunstore.ActivationStatusWaiting {
		return CallbackPayloadRead{}, ErrCallbackNotWaiting
	}
	section := strings.TrimSpace(command.Section)
	var raw []byte
	encoding := "json"
	switch section {
	case "input":
		raw = append([]byte(nil), snapshot.activation.InputJSON...)
	case "output_contract":
		raw, err = json.Marshal(snapshot.request.Intent.CallbackOutput)
	case "work_items_contract":
		value, valueErr := callbackResultFromSnapshot(snapshot, true)
		if valueErr != nil {
			return CallbackPayloadRead{}, valueErr
		}
		if value.WorkItemsContract == nil {
			return CallbackPayloadRead{}, storeerr.BadRequest("callback does not declare a workItems contract")
		}
		raw, err = json.Marshal(value.WorkItemsContract)
	default:
		return CallbackPayloadRead{}, storeerr.BadRequest("callback payload section is invalid")
	}
	if err != nil {
		return CallbackPayloadRead{}, fmt.Errorf("encode callback payload section: %w", err)
	}
	if section == "input" {
		if artifactPage, handled, artifactErr := service.readArtifactCallbackPage(ctx, snapshot, command, raw); handled {
			if artifactErr != nil {
				return CallbackPayloadRead{}, artifactErr
			}
			return artifactPage, nil
		}
	}
	if len(bytes.TrimSpace(raw)) == 0 {
		return CallbackPayloadRead{}, storeerr.BadRequest("callback payload section is empty")
	}
	contentDigest := workflowpayload.ContentDigest(raw)
	path := strings.TrimSpace(command.Path)
	if path != "" {
		raw, err = workflowpayload.NarrowJSON(raw, path)
		if err != nil {
			return CallbackPayloadRead{}, storeerr.BadRequest(err.Error())
		}
	}
	page, err := workflowpayload.PageBytesWithSnapshot(
		workflowpayload.Identity{Kind: "step", WorkspaceID: command.WorkspaceID, PlanID: snapshot.run.PlanID, RunID: command.RunID, StepID: command.CallbackID},
		section, path, encoding, raw, workflowpayload.Snapshot{
			Digest: contentDigest, ValueDigest: workflowpayload.ContentDigest(raw),
		}, strings.TrimSpace(command.Cursor), command.Limit,
	)
	if err != nil {
		if errors.Is(err, workflowpayload.ErrContentChanged) {
			return CallbackPayloadRead{}, fmt.Errorf("%w: restart the read: %w", ErrCallbackPayloadChanged, storeerr.ErrConflict)
		}
		return CallbackPayloadRead{}, storeerr.BadRequest(err.Error())
	}
	return CallbackPayloadRead{
		RunID: command.RunID, CallbackID: command.CallbackID,
		Status: snapshot.effect.Status, RunStatus: snapshot.run.Status, Page: page,
	}, nil
}

func (service *CallbackService) readArtifactCallbackPage(ctx context.Context, snapshot callbackSnapshot, command CallbackCommand, raw []byte) (CallbackPayloadRead, bool, error) {
	path, err := runartifact.PathSegments(command.Path)
	if err != nil {
		return CallbackPayloadRead{}, false, storeerr.BadRequest(err.Error())
	}
	ref, artifactPath, marker, err := runartifact.FindMarkerAtPath(raw, path)
	if err != nil || !marker {
		return CallbackPayloadRead{}, marker, err
	}
	if ref.WorkspaceID != snapshot.run.WorkspaceID || ref.RunID != snapshot.run.ID {
		return CallbackPayloadRead{}, true, ErrCallbackUnauthorized
	}
	reader, ok := service.repository.(CallbackArtifactReader)
	if !ok {
		return CallbackPayloadRead{}, true, ErrCallbackUnauthorized
	}
	page, err := reader.ReadArtifactPage(ctx, ref, command.Section, artifactPath, strings.TrimSpace(command.Cursor), command.Limit)
	if err != nil {
		return CallbackPayloadRead{}, true, err
	}
	return CallbackPayloadRead{
		RunID: command.RunID, CallbackID: command.CallbackID,
		Status: snapshot.effect.Status, RunStatus: snapshot.run.Status,
		Page: workflowpayload.Page{
			Section: command.Section, Path: runartifact.CanonicalPath(artifactPath), Encoding: "json", Value: string(page.Bytes),
			Cursor: command.Cursor, NextCursor: page.NextCursor, Bytes: len(page.Bytes), TotalBytes: page.TotalBytes,
			Complete: page.NextCursor == "",
		},
	}, true, nil
}

func (service *CallbackService) ExecuteCallback(ctx context.Context, command CallbackCommand) (CallbackResult, error) {
	if service == nil || service.repository == nil || service.now == nil {
		return CallbackResult{}, fmt.Errorf("workflow Run callback service is not configured")
	}
	command = normalizeCallbackCommand(command)
	if err := validateCallbackCommand(command); err != nil {
		return CallbackResult{}, err
	}
	now := service.now().UTC().Truncate(time.Microsecond)
	snapshot, err := service.loadSnapshot(ctx, command, now)
	if err != nil {
		return CallbackResult{}, err
	}
	if snapshot.receipt.CallbackResponderAgentID != command.AgentID {
		return CallbackResult{}, ErrCallbackUnauthorized
	}
	base, err := callbackResultFromSnapshot(snapshot, false)
	if err != nil {
		return CallbackResult{}, err
	}

	if command.Action == CallbackActionGetCurrent || command.Action == CallbackActionPause || command.Action == CallbackActionPark {
		return service.executeNonTerminalCallbackAction(ctx, command, snapshot, now)
	}

	terminalResult, activationOutput, err := buildCallbackCompletion(snapshot, command, now)
	if err != nil {
		return CallbackResult{}, err
	}
	if snapshot.effect.Status == workflowrunstore.EffectStatusCompleted {
		if callbackCompletionMatches(snapshot.receipt, terminalResult) {
			base.Replayed = true
			return base, nil
		}
		return CallbackResult{}, ErrCallbackAlreadyResolved
	}
	if snapshot.run.Terminal() {
		return CallbackResult{}, ErrRunTerminal
	}
	if snapshot.effect.Status != workflowrunstore.EffectStatusAwaiting || snapshot.activation.Status != workflowrunstore.ActivationStatusWaiting {
		return CallbackResult{}, ErrCallbackNotWaiting
	}
	terminalResult, activationOutput, scopeCandidates, err := service.normalizeCallbackScopeOutputs(
		snapshot,
		command,
		terminalResult,
		activationOutput,
	)
	if err != nil {
		return CallbackResult{}, err
	}

	effectResultJSON, activationOutputJSON, err := normalizedCallbackPersistencePayloads(
		terminalResult,
		activationOutput,
	)
	if err != nil {
		return CallbackResult{}, err
	}
	eventPayload, err := callbackCompletionEventPayload(snapshot, terminalResult, command.AgentID)
	if err != nil {
		return CallbackResult{}, fmt.Errorf("encode workflow Run callback event: %w", err)
	}
	resolved, err := service.repository.ResolveAwaitingEffect(ctx, workflowrunstore.ResolveAwaitingEffectParams{
		WorkspaceID: command.WorkspaceID, RunID: command.RunID,
		ActivationID: snapshot.activation.ID, EffectID: snapshot.effect.ID,
		AgentID: command.AgentID, MemberID: command.MemberID,
		EffectResultJSON: effectResultJSON, ActivationOutputJSON: activationOutputJSON,
		SelectedPort: terminalResult.Port, RunnableAt: now,
		Events: []workflowrunstore.EventDraft{{Type: "message_callback_completed", PayloadJSON: eventPayload}},
		Now:    now,
		PreCommitGuard: service.callbackScopePreCommitGuard(
			command,
			scopeCandidates,
			callbackAgentResolvedTaskWatchScope(snapshot.request.Intent, terminalResult),
		),
	})
	if err != nil {
		if errors.Is(err, storeerr.ErrConflict) {
			return service.resolveConcurrentReplay(ctx, command, terminalResult, now)
		}
		if errors.Is(err, storeerr.ErrNotFound) {
			return CallbackResult{}, ErrCallbackNotFound
		}
		return CallbackResult{}, err
	}
	base.Status = resolved.Effect.Status
	base.RunStatus = resolved.Run.Status
	base.Replayed = resolved.Replayed
	return base, nil
}

// executeNonTerminalCallbackAction handles get_current/pause/park — callback
// operations that are NOT historical Run queries. The frozen workScope and
// relatedScopes are execution context, not fresh authorization requests; Agent
// callback ownership was already established when loading the snapshot.
// Once the Run/effect/activation is no longer waiting, it returns the same
// authoritative terminal signal as complete/fail so the Agent daemon can revoke
// that callback authority and release the next callback turn. Pause and park are
// durable, non-terminal FSM mutations whose losing branches map to the normalized
// not-waiting response.
func (service *CallbackService) executeNonTerminalCallbackAction(
	ctx context.Context,
	command CallbackCommand,
	snapshot callbackSnapshot,
	now time.Time,
) (CallbackResult, error) {
	if command.Action == CallbackActionPause {
		if snapshot.receipt.CallbackAction != "" || snapshot.receipt.CallbackActionIdempotencyKey != "" {
			if snapshot.receipt.CallbackAction == command.Action && snapshot.receipt.CallbackActionIdempotencyKey == command.IdempotencyKey {
				base, err := callbackResultFromSnapshot(snapshot, true)
				if err != nil {
					return CallbackResult{}, err
				}
				base.Replayed = true
				return base, nil
			}
			return CallbackResult{}, ErrCallbackReplayConflict
		}
	}
	if snapshot.run.Terminal() || snapshot.effect.Status != workflowrunstore.EffectStatusAwaiting ||
		snapshot.activation.Status != workflowrunstore.ActivationStatusWaiting {
		return CallbackResult{}, ErrCallbackNotWaiting
	}
	replayed := false
	switch command.Action {
	case CallbackActionPause:
		// pause is non-terminal: the effect stays awaiting and automatic
		// recovery is stopped durably and permanently. Only callback
		// complete/fail, cancel, or terminal lifecycle ends/cleans it.
		// Consume the repository boolean: a losing pause (the park FSM already
		// owns the effect, or the Run/effect moved) maps to the canonical
		// not-waiting response instead of a stale Applied from the pre-mutation
		// snapshot.
		var paused workflowrunstore.CallbackRecoveryActionResult
		var err error
		if command.MemberID != "" {
			mutationRepository, ok := service.repository.(CallbackRecoveryMutationRepository)
			if !ok {
				return CallbackResult{}, ErrCallbackUnauthorized
			}
			paused, err = mutationRepository.PauseAwaitingCallbackRecoveryForAgent(ctx, workflowrunstore.CallbackRecoveryMutationParams{
				WorkspaceID: command.WorkspaceID, RunID: command.RunID, EffectID: command.CallbackID,
				AgentID: command.AgentID, MemberID: command.MemberID, Now: now, IdempotencyKey: command.IdempotencyKey,
			})
		} else {
			paused, err = service.repository.PauseAwaitingCallbackRecoveryWithIdempotency(
				ctx, command.WorkspaceID, command.RunID, command.CallbackID, now, command.IdempotencyKey,
			)
		}
		if err != nil {
			if errors.Is(err, storeerr.ErrConflict) {
				return CallbackResult{}, ErrCallbackReplayConflict
			}
			return CallbackResult{}, err
		}
		if !paused.Applied {
			return CallbackResult{}, ErrCallbackNotWaiting
		}
		replayed = paused.Replayed
	case CallbackActionPark:
		// park is non-terminal: it arms the durable park FSM so recovery
		// stops while the responder is idle and resumes on a fresh schedule
		// after the first authoritative working observation.
		// Consume the repository boolean: a losing park (pause already won, or
		// the Run/effect moved) maps to the canonical not-waiting response.
		var armed workflowrunstore.CallbackRecoveryActionResult
		var err error
		if command.MemberID != "" {
			mutationRepository, ok := service.repository.(CallbackRecoveryMutationRepository)
			if !ok {
				return CallbackResult{}, ErrCallbackUnauthorized
			}
			armed, err = mutationRepository.ParkAwaitingCallbackRecoveryForAgent(ctx, workflowrunstore.CallbackRecoveryMutationParams{
				WorkspaceID: command.WorkspaceID, RunID: command.RunID, EffectID: command.CallbackID,
				AgentID: command.AgentID, MemberID: command.MemberID, Now: now,
			})
		} else {
			armed, err = service.repository.ParkAwaitingCallbackRecoveryWithIdempotency(
				ctx, command.WorkspaceID, command.RunID, command.CallbackID, now, "",
			)
		}
		if err != nil {
			if errors.Is(err, storeerr.ErrConflict) {
				return CallbackResult{}, ErrCallbackReplayConflict
			}
			return CallbackResult{}, err
		}
		if !armed.Applied {
			return CallbackResult{}, ErrCallbackNotWaiting
		}
		replayed = armed.Replayed
	}
	result, err := callbackResultFromSnapshot(snapshot, true)
	if err != nil {
		return CallbackResult{}, err
	}
	result.Replayed = replayed
	return result, nil
}

func normalizedCallbackPersistencePayloads(
	result EffectResult,
	output runkernel.Values,
) ([]byte, []byte, error) {
	effectResultJSON, err := EncodeEffectResult(result)
	if err != nil {
		return nil, nil, err
	}
	effectResultJSON, err = workflowrunstore.DeterministicPayloadJSON(
		"workflow Run callback effect result",
		effectResultJSON,
		true,
	)
	if err != nil {
		cause := fmt.Errorf("callback result exceeds the durable payload budget: %w", err)
		return nil, nil, callbackValidationError(
			CallbackValidationCodeOutputBudget,
			callbackDiagnosticsFromCause(cause),
			cause,
		)
	}
	activationOutputJSON, err := json.Marshal(output)
	if err != nil {
		return nil, nil, fmt.Errorf("encode workflow Run callback output: %w", err)
	}
	activationOutputJSON, err = workflowrunstore.DeterministicPayloadJSON(
		"workflow Run callback activation output",
		activationOutputJSON,
		true,
	)
	if err != nil {
		cause := fmt.Errorf("callback output exceeds the durable payload budget: %w", err)
		return nil, nil, callbackValidationError(
			CallbackValidationCodeOutputBudget,
			callbackDiagnosticsFromCause(cause),
			cause,
		)
	}
	return effectResultJSON, activationOutputJSON, nil
}

// callbackCompletionEventPayload builds the durable message_callback_completed
// event payload. When the callback terminated with a failure, it carries the
// canonical safe failure contract (code/stage/path/problem/hint/retryable/
// diagnostics) so the Run log Details projection preserves every field the
// Agent/CLI needs; the raw responder failure message never enters the event.
func callbackCompletionEventPayload(
	snapshot callbackSnapshot,
	result EffectResult,
	agentID string,
) ([]byte, error) {
	payload := map[string]any{
		"effectId": snapshot.effect.ID, "activationId": snapshot.activation.LogicalKey,
		"nodeId": snapshot.activation.NodeID, "agentId": agentID,
		"port": result.Port,
	}
	if result.FailureCode != "" {
		payload["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.
		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
		}
	}
	return json.Marshal(payload)
}

func (service *CallbackService) resolveConcurrentReplay(ctx context.Context, command CallbackCommand, expected EffectResult, now time.Time) (CallbackResult, error) {
	snapshot, err := service.loadSnapshot(ctx, command, now)
	if err != nil {
		return CallbackResult{}, err
	}
	if snapshot.receipt.CallbackResponderAgentID != command.AgentID {
		return CallbackResult{}, ErrCallbackUnauthorized
	}
	if snapshot.effect.Status == workflowrunstore.EffectStatusCompleted {
		if !callbackCompletionMatches(snapshot.receipt, expected) {
			return CallbackResult{}, ErrCallbackAlreadyResolved
		}
		result, resultErr := callbackResultFromSnapshot(snapshot, false)
		if resultErr != nil {
			return CallbackResult{}, resultErr
		}
		result.Replayed = true
		return result, nil
	}
	if snapshot.run.Terminal() {
		return CallbackResult{}, ErrRunTerminal
	}
	return CallbackResult{}, ErrCallbackNotWaiting
}

func (service *CallbackService) loadSnapshot(ctx context.Context, command CallbackCommand, now time.Time) (callbackSnapshot, error) {
	if err := service.authorizeCallbackCommand(ctx, command); err != nil {
		return callbackSnapshot{}, err
	}
	run, err := service.repository.LoadRun(ctx, command.WorkspaceID, command.RunID)
	if err != nil {
		return callbackSnapshot{}, mapCallbackReadError(err)
	}
	effect, err := service.repository.LoadEffect(ctx, command.WorkspaceID, command.RunID, command.CallbackID)
	if err != nil {
		return callbackSnapshot{}, mapCallbackReadError(err)
	}
	activation, err := service.repository.LoadActivation(ctx, command.WorkspaceID, command.RunID, effect.ActivationID)
	if err != nil {
		return callbackSnapshot{}, mapCallbackReadError(err)
	}
	return callbackSnapshotFromRecords(run, activation, effect, now)
}

func callbackSnapshotFromRecords(
	run workflowrunstore.Run,
	activation workflowrunstore.Activation,
	effect workflowrunstore.Effect,
	now time.Time,
) (callbackSnapshot, error) {
	request, err := DecodeEffectRequest(effect.RequestJSON)
	if err != nil {
		return callbackSnapshot{}, fmt.Errorf("decode workflow Run callback request: %w", err)
	}
	if effect.Kind != EffectKindMessage || request.Intent.Kind != runkernel.IntentSendMessage || !request.Intent.Callback || request.Message == nil ||
		request.Intent.ActivationID != activation.LogicalKey || request.Intent.NodeID != activation.NodeID || request.Message.EffectID != effect.ID {
		return callbackSnapshot{}, ErrCallbackNotFound
	}
	resultJSON := bytes.TrimSpace(effect.ResultJSON)
	if len(resultJSON) == 0 {
		// WriteRunMessage wakes the Agent before the dispatcher can atomically
		// advance dispatching -> awaiting and persist the delivery receipt plus
		// separately frozen callback authority. A
		// replay-receipt lookup can also fail after the message commit, moving the
		// effect back to pending while that Agent already owns the hidden gate.
		// Both live states are retriable until the receipt converges or the effect
		// reaches a definitive status.
		if callbackReceiptMayStillConverge(run, activation, effect, request) {
			return callbackSnapshot{}, ErrCallbackNotReady
		}
		return callbackSnapshot{}, ErrCallbackNotWaiting
	}
	// Failed, canceled, delivered, or otherwise non-awaiting callback effects do
	// not carry an authorization receipt. Their ResultJSON may be a dispatcher
	// failure result rather than a delivery receipt; treat that durable state as
	// authoritative terminal evidence instead of surfacing receipt corruption.
	if effect.Status != workflowrunstore.EffectStatusAwaiting && effect.Status != workflowrunstore.EffectStatusCompleted {
		return callbackSnapshot{}, ErrCallbackNotWaiting
	}
	receipt, err := DecodeEffectResult(effect.ResultJSON)
	if err != nil {
		return callbackSnapshot{}, fmt.Errorf("decode workflow Run callback receipt: %w", err)
	}
	// Omission remains readable for receipts committed before deliveryScope was
	// introduced. New writers require the field; whenever it is present here it
	// must be canonical before get_current or replay can expose the receipt.
	if receipt.DeliveryScope != nil {
		scope, err := domain.NormalizeMessageScopeValue(*receipt.DeliveryScope)
		if err != nil {
			return callbackSnapshot{}, fmt.Errorf("workflow Run callback receipt delivery scope is invalid: %w", err)
		}
		receipt.DeliveryScope = &scope
	}
	if receipt.WorkScope != nil {
		scope, err := domain.NormalizeMessageScopeValue(*receipt.WorkScope)
		if err != nil {
			return callbackSnapshot{}, fmt.Errorf("workflow Run callback receipt work scope is invalid: %w", err)
		}
		receipt.WorkScope = &scope
	}
	if strings.TrimSpace(receipt.MessageID) == "" {
		if effect.Status == workflowrunstore.EffectStatusCompleted {
			// A completed timeout/failure result is definitive but is not a message
			// delivery receipt. Valid completed callback receipts still flow below
			// and retain exact replay/already-resolved authorization semantics.
			return callbackSnapshot{}, ErrCallbackNotWaiting
		}
		return callbackSnapshot{}, fmt.Errorf("workflow Run callback receipt is incomplete")
	}
	receipt.CallbackResponderAgentID, err = ResolveCallbackResponderAuthority(
		request.Intent.Callback,
		receipt.CallbackResponderAgentID,
		receipt.WakeAgentIDs,
	)
	if err != nil {
		return callbackSnapshot{}, ErrCallbackUnauthorized
	}
	return callbackSnapshot{run: run, activation: activation, effect: effect, request: request, receipt: receipt}, nil
}

func callbackReceiptMayStillConverge(
	run workflowrunstore.Run,
	activation workflowrunstore.Activation,
	effect workflowrunstore.Effect,
	request EffectRequest,
) bool {
	if run.Terminal() || activationTerminal(activation.Status) {
		return false
	}
	switch effect.Status {
	case workflowrunstore.EffectStatusPending, workflowrunstore.EffectStatusDispatching:
		return true
	default:
		return false
	}
}

func activationTerminal(status string) bool {
	switch status {
	case workflowrunstore.ActivationStatusSucceeded, workflowrunstore.ActivationStatusFailed, workflowrunstore.ActivationStatusCanceled:
		return true
	default:
		return false
	}
}

func callbackResultFromSnapshot(snapshot callbackSnapshot, includeCollaborationScopes bool) (CallbackResult, error) {
	contractJSON, err := json.Marshal(snapshot.request.Intent.CallbackOutput)
	if err != nil {
		return CallbackResult{}, fmt.Errorf("encode workflow Run callback output contract: %w", err)
	}
	input, inputRef, err := workflowpayload.InlineWithSnapshot(workflowpayload.Identity{Kind: "step", WorkspaceID: snapshot.run.WorkspaceID, PlanID: snapshot.run.PlanID, RunID: snapshot.run.ID, StepID: snapshot.effect.ID}, "input", "", "json", snapshot.activation.InputJSON, workflowpayload.Snapshot{})
	if err != nil {
		return CallbackResult{}, fmt.Errorf("bound callback input: %w", err)
	}
	contract, contractRef, err := workflowpayload.InlineWithSnapshot(workflowpayload.Identity{Kind: "step", WorkspaceID: snapshot.run.WorkspaceID, PlanID: snapshot.run.PlanID, RunID: snapshot.run.ID, StepID: snapshot.effect.ID}, "output_contract", "", "json", contractJSON, workflowpayload.Snapshot{})
	if err != nil {
		return CallbackResult{}, fmt.Errorf("bound callback output contract: %w", err)
	}
	result := CallbackResult{
		RunID: snapshot.run.ID, CallbackID: snapshot.effect.ID,
		ActivationID: snapshot.activation.LogicalKey, NodeID: snapshot.activation.NodeID,
		Status: snapshot.effect.Status, RunStatus: snapshot.run.Status,
		Input: append(json.RawMessage(nil), input...), OutputContract: append(json.RawMessage(nil), contract...),
		InputRef: inputRef, OutputContractRef: contractRef,
	}
	if snapshot.receipt.DeliveryScope != nil {
		scope := *snapshot.receipt.DeliveryScope
		result.DeliveryScope = &scope
	}
	// The minimal member context is already frozen on the durable effect
	// receipt; surface it verbatim so callback/retry/replay reuse the same
	// snapshot.
	result.TeamMemberContext = cloneCallbackTeamMemberContext(snapshot.receipt.TeamMemberContext)
	if includeCollaborationScopes && snapshot.request.Intent.WorkScopeMode != "" {
		switch snapshot.request.Intent.WorkScopeMode {
		case runkernel.WorkScopeModeAgent:
			result.WorkScope = json.RawMessage("null")
		case runkernel.WorkScopeModeFixed:
			if snapshot.request.Intent.WorkScope == nil {
				return CallbackResult{}, fmt.Errorf("workflow Run callback fixed workScope is missing")
			}
			workScopeJSON, err := json.Marshal(snapshot.request.Intent.WorkScope)
			if err != nil {
				return CallbackResult{}, fmt.Errorf("encode workflow Run callback workScope: %w", err)
			}
			result.WorkScope = workScopeJSON
		default:
			return CallbackResult{}, fmt.Errorf("workflow Run callback workScope mode is invalid")
		}
		// The authored collaboration field is always present on the callback response,
		// including when the frozen list is empty. Start from a non-nil slice so
		// the pointer encodes as [] rather than null.
		relatedScopes := append([]domain.MessageScopeValue{}, snapshot.request.Intent.RelatedScopes...)
		result.RelatedScopes = &relatedScopes
	}
	if includeCollaborationScopes && snapshot.request.Intent.WorkItemsAllowedResponders != nil {
		if snapshot.request.Intent.WorkItemsDependencyValueContract == nil {
			return CallbackResult{}, fmt.Errorf("workflow Run WorkItems callback dependency value contract is missing")
		}
		dependencyValueContract := append(
			plan.ArtifactContract{},
			(*snapshot.request.Intent.WorkItemsDependencyValueContract)...,
		)
		workItemsContract := &CallbackWorkItemsContract{
			Required:                  true,
			MaxItems:                  plan.MaxWorkItems,
			MaxKeyBytes:               plan.MaxWorkItemKeyBytes,
			MaxRelatedScopesPerItem:   plan.MaxSendMessageRelatedScopes,
			MaxDependencyEdges:        plan.MaxWorkItemDependencyEdges,
			MaxRelatedScopeReferences: plan.MaxWorkItemRelatedScopeReferences,
			MaxJSONValues:             plan.MaxWorkItemsJSONValues,
			MaxJSONBytes:              plan.MaxWorkItemsJSONBytes,
			AllowedResponders:         append([]string{}, snapshot.request.Intent.WorkItemsAllowedResponders...),
			DependencyValueContract:   dependencyValueContract,
		}
		encodedWorkItemsContract, marshalErr := json.Marshal(workItemsContract)
		if marshalErr != nil {
			return CallbackResult{}, fmt.Errorf("encode workflow Run WorkItems contract: %w", marshalErr)
		}
		inlineWorkItemsContract, workItemsRef, inlineErr := workflowpayload.InlineWithSnapshot(workflowpayload.Identity{Kind: "step", WorkspaceID: snapshot.run.WorkspaceID, PlanID: snapshot.run.PlanID, RunID: snapshot.run.ID, StepID: snapshot.effect.ID}, "work_items_contract", "", "json", encodedWorkItemsContract, workflowpayload.Snapshot{})
		if inlineErr != nil {
			return CallbackResult{}, fmt.Errorf("bound workflow Run WorkItems contract: %w", inlineErr)
		}
		if workItemsRef == nil {
			if err := json.Unmarshal(inlineWorkItemsContract, workItemsContract); err != nil {
				return CallbackResult{}, fmt.Errorf("decode bounded workflow Run WorkItems contract: %w", err)
			}
			result.WorkItemsContract = workItemsContract
		} else {
			result.WorkItemsRef = workItemsRef
		}
	}
	return result, nil
}

func buildCallbackCompletion(snapshot callbackSnapshot, command CallbackCommand, now time.Time) (EffectResult, runkernel.Values, error) {
	result := snapshot.receipt.Normalize()
	result.IdempotencyKey = command.IdempotencyKey
	result.CallbackAction = ""
	result.CallbackActionIdempotencyKey = ""
	result.CompletedAt = &now
	result.FailureCode, result.FailureMessage = "", ""
	var output runkernel.Values
	switch command.Action {
	case CallbackActionComplete:
		payload := command.Output
		if len(bytes.TrimSpace(payload)) == 0 {
			payload = json.RawMessage(`{}`)
		}
		decoded, err := decodeCallbackValues(payload)
		if err != nil {
			cause := errors.Join(err)
			return EffectResult{}, nil, callbackValidationError(
				CallbackValidationCodeOutputContract,
				callbackDiagnosticsFromCause(cause),
				cause,
			)
		}
		// Validate the independent dimensions (output contract, workItems,
		// workScope) and collect every diagnostic, so a single complete attempt
		// surfaces all typed validation problems at once instead of the first
		// failure only.
		var diagnostics []publicerr.Diagnostic
		var causes []error
		normalized, err := runkernel.NormalizeContract("callback.output", snapshot.request.Intent.CallbackOutput, decoded)
		if err != nil {
			diagnostics = append(diagnostics, callbackDiagnosticsFromCause(err)...)
			causes = append(causes, err)
		} else {
			output = normalized
		}
		workItems, workItemsErr := callbackCompletionWorkItems(snapshot.request.Intent, command.WorkItems)
		if workItemsErr != nil {
			diagnostics = append(diagnostics, callbackDiagnosticsFromError(workItemsErr)...)
			causes = append(causes, workItemsErr)
		}
		workScope, workScopeErr := callbackCompletionWorkScope(snapshot.request.Intent, command.WorkScope)
		if workScopeErr != nil {
			diagnostics = append(diagnostics, callbackDiagnosticsFromError(workScopeErr)...)
			causes = append(causes, workScopeErr)
		}
		if len(diagnostics) != 0 {
			firstCode := ""
			if len(causes) != 0 {
				firstCode = callbackValidationCodeFromError(causes[0])
			}
			return EffectResult{}, nil, mergeCallbackValidationDiagnostics(
				CallbackValidationCodeOutputContract,
				diagnostics,
				firstCode,
				errors.Join(causes...),
			)
		}
		if workItems != nil {
			output = cloneCallbackValues(output)
			output["workItems"] = runkernel.WorkItemsValue(workItems)
		}
		if workScope != nil {
			output = cloneCallbackValues(output)
			output["workScope"] = callbackScopeMap(*workScope)
			result.WorkScope = workScope
		}
		result.Port = callbackCompletedPort
		result.Output = output
	case CallbackActionFail:
		output = runkernel.Values{}
		result.Port = callbackFailedPort
		result.Output = output
		result.FailureCode = command.FailureCode
		result.FailureStage = "callback"
		// The responder's arbitrary FailureMessage is never persisted: the stable
		// code→problem/hint mapping is the durable public contract, and the raw
		// message stays in the callback command only. This mirrors the dispatcher
		// terminal failure path so no unreviewed free text reaches the durable
		// effect ResultJSON or StateJSON.
		result.FailureProblem = runkernel.FailureProblem(command.FailureCode)
		result.FailureHint = runkernel.FailureHint(command.FailureCode)
		result.FailureMessage = runkernel.FailureHint(command.FailureCode)
		result.FailureDiagnostics = nil
	default:
		return EffectResult{}, nil, fmt.Errorf("unsupported workflow Run callback action %q", command.Action)
	}
	return result.Normalize(), output, nil
}

var _ CallbackRuntime = (*CallbackService)(nil)