package orchestrator

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"strconv"
	"strings"
	"time"

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

// recoveryPromptText is the authoritative English Recovery Prompt frozen by the
// Leader (2026-08-12). It is rendered with real Run/Step selectors and the
// 1-based attempt index; the receiving Agent must be able to copy the exact
// command into the public `loop exec workflow step` CLI. The blank line before
// the attempt line is part of the frozen text.
const recoveryPromptText = "The Workflow step you were executing is still waiting for its callback, but you have stopped working. Continue the step. First run `loop exec workflow step '{\"action\":\"get\",\"runId\":\"<run-id>\",\"stepId\":\"<step-id>\"}'` to inspect the active callback and output contract. Use the trusted Run and Step selectors included with this recovery delivery. Check the work and artifacts already produced; do not restart without reason. Call `complete` when finished, `fail` if the step cannot be completed, `pause` if you need to stop automatic recovery permanently while you continue discussing with the user (only completing, failing, or canceling the step ends the pause), or `park` if you want automatic recovery to stop while you are idle and then resume on a fresh schedule once you start working again. An ordinary chat reply does not advance or pause the Workflow.\n\nThis is automatic recovery attempt {{recoveryIndex}} of 5."

// renderRecoveryPrompt materializes the frozen Recovery Prompt with the exact
// trusted Run and Step selectors and the 1-based attempt index. The command
// presentation matches the existing workflow prompt compiler
// (`workflowContextPrompt`): `loop exec workflow step '{"action":"get",
// "runId":"<run-id>","stepId":"<step-id>"}'`. The model is never asked to guess
// the selectors.
func renderRecoveryPrompt(runID, stepID string, attemptIndex int) string {
	content := strings.ReplaceAll(recoveryPromptText, "<run-id>", strings.TrimSpace(runID))
	content = strings.ReplaceAll(content, "<step-id>", strings.TrimSpace(stepID))
	content = strings.ReplaceAll(content, "{{recoveryIndex}}", strconv.Itoa(attemptIndex))
	return content
}

// recoveryClientMessageID is a deterministic per-(effect, delivery sequence)
// client message id. The delivery sequence is the durable, monotonic send
// history on the effect—not the resettable current-cycle display index—so a
// recovery send remains unique across Agent recovery resets while a retry of
// the same committed send reuses its id. It intentionally differs from the
// original callback delivery's client message id.
func recoveryClientMessageID(runID, effectID string, deliverySequence int) string {
	return fmt.Sprintf("workflow-run-recovery:%s:%s:%d", runID, effectID, deliverySequence)
}

// RecoveryRepository is the smallest persistence boundary needed by the
// automatic recovery pass. It scans awaiting send_message.callback effects and
// atomically advances one effect's durable recovery state.
type RecoveryRepository interface {
	ListRecoveryCandidates(context.Context, run.ListRecoveryCandidatesParams) ([]run.Effect, error)
	RecoveryTick(context.Context, run.RecoveryTickParams) (run.RecoveryTickResult, error)
}

// RuntimeWorkingReader reports the authoritative Agent runtime working state for a
// frozen callback responder. It MUST query the same real Agent turn/execution
// state that drives Web Agent Activity and daemon scheduling (the daemon/Hub
// overlay), never the Workflow capacity lease or callback FIFO.
//
// working=true means the responder has a real active turn or a real runtime
// queued turn — recovery must not fire and no timing is started. known=false
// means the current state cannot be determined (for example the machine is not
// connected, the Hub has not reconnected after a server/worker restart, or the
// responder is within the bounded snapshot grace); the tick must NOT send and
// must NOT start recovery timing, and must never fall back to the Workflow
// capacity lease/FIFO.
type RuntimeWorkingReader interface {
	RuntimeWorking(context.Context, string, string) (working bool, known bool)
}

// RuntimeObservationReader is the production authoritative runtime observation
// reader. It resolves the responder's machine, acquires a machine observation
// guard that prevents authoritative turn-state replacement while held, and
// samples the immutable observation (including the derived working/known
// judgment). The caller MUST call the returned release AFTER the row-locked
// RecoveryTick transaction commits, and on every error/unknown path. ok=false
// means the observation could not be acquired (unknown/offline) and no guard is
// held. A reader that does not implement this interface degrades to a plain
// working/known readout with no guard (legacy/test wiring only; production
// always uses the observation reader).
type RuntimeObservationReader interface {
	AcquireRuntimeObservation(context.Context, string, string) (run.RecoveryObservation, func(), bool)
}

// RuntimeWorkingReasonReader is an optional richer readout a RuntimeWorkingReader
// may implement for observability. It returns a compact reason for the judgment:
// "working", "idle", "offline", "grace", or "unknown". It is best-effort and
// never gates the recovery decision.
type RuntimeWorkingReasonReader interface {
	RuntimeWorkingReason(context.Context, string, string) string
}

// RecoveryOutcome is the bounded per-pass accounting surfaced for logging and
// tests.
type RecoveryOutcome struct {
	Candidates int
	Working    int
	Unknown    int
	Sent       int
	Failed     int
}

type RecoveryRunnerOptions struct {
	WorkerID         string
	WorkspaceGroupID string
	Limit            int
	Now              func() time.Time
	// Logger is optional; when set, recovery skips due to an unknown runtime
	// state (disconnect / runtime exit / old daemon) are logged so the fail-safe
	// no-send behavior is observable.
	Logger *slog.Logger
	// RuntimeWorking is the authoritative Agent runtime working-state reader. It is
	// required: recovery MUST judge working from the real runtime state.
	RuntimeWorking RuntimeWorkingReader
}

// RecoveryRunner is the automatic recovery worker pass for awaiting
// send_message.callback effects whose frozen responder stopped working. It is a
// separate runner with its own bounded scan; it never wakes the reducer, never
// sets runnableAt, never releases the execution lease, and never pokes another
// Run. Each recovery nudge is an ordinary send_message to the frozen delivery
// scope that reuses the frozen message request facts.
type RecoveryRunner struct {
	repository RecoveryRepository
	messages   MessageWriter
	options    RecoveryRunnerOptions
}

func NewRecoveryRunner(
	repository RecoveryRepository,
	messages MessageWriter,
	options RecoveryRunnerOptions,
) (*RecoveryRunner, error) {
	if repository == nil {
		return nil, fmt.Errorf("workflow Run recovery repository is required")
	}
	if messages == nil {
		return nil, fmt.Errorf("workflow Run recovery message writer is required")
	}
	if options.RuntimeWorking == nil {
		return nil, fmt.Errorf("workflow Run recovery runtime working reader is required")
	}
	options.WorkerID = strings.TrimSpace(options.WorkerID)
	if options.WorkerID == "" || len(options.WorkerID) > 128 {
		return nil, fmt.Errorf("workflow Run recovery workerId is required and must be bounded")
	}
	if options.Limit == 0 {
		options.Limit = defaultEffectClaimLimit
	}
	if options.Limit < 1 || options.Limit > runtimeCollectionPage {
		return nil, fmt.Errorf("workflow Run recovery limit must be between 1 and %d", runtimeCollectionPage)
	}
	if options.Now == nil {
		options.Now = time.Now
	}
	return &RecoveryRunner{repository: repository, messages: messages, options: options}, nil
}

// RecoverOnce scans awaiting callbacks and performs exactly the due recovery
// sends for one bounded pass. For each candidate the runner first asks the
// authoritative runtime working reader. Unknown/offline states are skipped without
// sending and without any durable write (fail-safe); every known observation —
// working OR idle — reaches the store's one row-locked conditional transition so
// the park FSM and the ordinary schedule share a single winner. Concurrent passes
// cannot double-send an effect: RecoveryTick persists the current-cycle display
// state and monotonic delivery sequence under the effect's row lock, so only
// one pass wins each scheduled slot.
func (runner *RecoveryRunner) RecoverOnce(ctx context.Context) (RecoveryOutcome, error) {
	if runner == nil || runner.repository == nil || runner.messages == nil || runner.options.RuntimeWorking == nil {
		return RecoveryOutcome{}, fmt.Errorf("workflow Run recovery runner is not configured")
	}
	scanAt := runner.clockNow()
	candidates, err := runner.repository.ListRecoveryCandidates(ctx, run.ListRecoveryCandidatesParams{
		WorkspaceGroupID: runner.options.WorkspaceGroupID, Now: scanAt, Limit: runner.options.Limit,
	})
	if err != nil {
		return RecoveryOutcome{}, err
	}
	outcome := RecoveryOutcome{Candidates: len(candidates)}
	var failures []error
	for _, effect := range candidates {
		// processCandidate owns the authoritative runtime observation lease for
		// this candidate: it acquires the machine observation guard and defers the
		// release immediately, so every success, unknown, repository error,
		// context cancellation, and panic unwinding path releases exactly once.
		runner.processCandidate(ctx, effect, &outcome, &failures)
	}
	return outcome, errors.Join(failures...)
}

// processCandidate handles one recovery candidate. The observation lease is
// owned ONLY by the inner tickWithObservationLease scope, which acquires it,
// defers the release, invokes only the row-locked RecoveryTick, and returns the
// durable result/working/reason. The outer handler performs the recovery message
// send ONLY after that inner helper has returned and the lease is released, so
// no external message/fanout I/O ever runs while the machine observation guard
// is held.
func (runner *RecoveryRunner) processCandidate(ctx context.Context, effect run.Effect, outcome *RecoveryOutcome, failures *[]error) {
	tick, unknown := runner.tickWithObservationLease(ctx, effect)
	if unknown {
		if runner.options.Logger != nil {
			reason := tick.reason
			if reason == "" {
				reason = runner.runtimeWorkingReason(ctx, effect)
			}
			runner.options.Logger.Debug("workflow Run recovery skipped unknown runtime state",
				slog.String("effect_id", effect.ID),
				slog.String("run_id", effect.RunID),
				slog.String("responder", effect.CallbackResponderAgentID),
				slog.String("reason", reason),
			)
		}
		outcome.Unknown++
		return
	}
	if tick.err != nil {
		outcome.Failed++
		*failures = append(*failures, fmt.Errorf("recovery tick workflow Run effect %s: %w", effect.ID, tick.err))
		return
	}
	if tick.working {
		outcome.Working++
		return
	}
	if !tick.send {
		return
	}
	// The observation lease was released by tickWithObservationLease before this
	// point, so the send runs outside the machine observation guard.
	if err := runner.sendRecoveryMessage(ctx, tick.effect, tick.displayAttemptIndex); err != nil {
		outcome.Failed++
		*failures = append(*failures, fmt.Errorf("send workflow Run recovery for effect %s: %w", effect.ID, err))
		return
	}
	outcome.Sent++
}

// candidateTickResult is the durable result of one row-locked RecoveryTick plus
// the observation judgment that drove it. displayAttemptIndex is resettable
// presentation state; the monotonic delivery sequence remains on effect. It is
// returned from the inner lease scope so the outer handler can send only after
// the lease is released.
type candidateTickResult struct {
	effect              run.Effect
	send                bool
	displayAttemptIndex int
	working             bool
	reason              string
	err                 error
}

// tickWithObservationLease owns the authoritative runtime observation lease for
// one candidate and performs ONLY the row-locked RecoveryTick under it. It
// acquires the machine observation guard, defers the release immediately after
// acquisition, and invokes only the row-locked tick inside, so the lease is held
// for exactly observation + RecoveryTick commit/error and released on every path
// (success, unknown, repository error, context cancellation, panic unwinding)
// exactly once. The returned candidateTickResult carries the durable tick result
// and the working/reason judgment for the outer handler.
func (runner *RecoveryRunner) tickWithObservationLease(ctx context.Context, effect run.Effect) (candidateTickResult, bool) {
	observation, release, ok := runner.acquireRuntimeObservation(ctx, effect)
	if release != nil {
		// Own the lease for the whole tick: defer the release so every path out of
		// this inner scope (unknown, working, idle send decision, repository error,
		// cancellation, panic) releases it exactly once. The release is idempotent
		// (sync.Once), so a double-call is safe, but the defer guarantees the single
		// intended call.
		defer release()
	}
	if !ok || !observation.Known {
		// unknown (machine not connected / Hub not reconnected after restart /
		// within bounded snapshot grace / old daemon without turn-state
		// reporting / offline responder): fail-safe — never send, never start
		// timing, zero transition/write.
		return candidateTickResult{reason: observation.Reason}, true
	}
	// Both known working and known idle observations reach the one row-locked
	// conditional transition. Working observations keep the ordinary schedule
	// untouched but can clear a parked effect; idle observations drive the
	// ordinary 1/2/4/8/16 cadence and confirm park.
	//
	// Authoritative fencing: the observation was sampled under a machine
	// observation guard (acquired by acquireRuntimeObservation) that prevents
	// authoritative turn-state replacement while held. The guard is kept valid
	// until RecoveryTick's row-locked transaction commits, then released by the
	// defer above on every path. The store never re-reads runtime state or
	// invokes any Hub/machine-resolution callback inside the transaction, so a
	// bare boolean sampled before the transaction is never sufficient and there
	// is no DB->Hub or Hub->DB lock inversion inside the tick.
	result, err := runner.repository.RecoveryTick(ctx, run.RecoveryTickParams{
		WorkspaceID: effect.WorkspaceID, RunID: effect.RunID, EffectID: effect.ID,
		Now: runner.clockNow(), Observation: observation,
	})
	if err != nil {
		return candidateTickResult{err: err}, false
	}
	return candidateTickResult{
		effect:              result.Effect,
		send:                result.Send,
		displayAttemptIndex: result.DisplayAttemptIndex,
		working:             observation.Working,
		reason:              observation.Reason,
	}, false
}

// acquireRuntimeObservation obtains the immutable authoritative runtime
// observation for one callback responder. When the injected reader implements
// RuntimeObservationReader (production: HubWorkingReader), it resolves the
// machine, acquires a machine observation guard, and samples the observation;
// the returned release MUST be called after the row-locked RecoveryTick
// transaction commits (and on every path). When the reader is a plain
// RuntimeWorkingReader (legacy/test wiring), it synthesizes a best-effort
// observation with no guard (release is nil). A missing frozen responder or any
// reader-side failure resolves to unknown (ok=false, known=false) with no guard
// held, so a transient reader failure never triggers a recovery send.
func (runner *RecoveryRunner) acquireRuntimeObservation(ctx context.Context, effect run.Effect) (run.RecoveryObservation, func(), bool) {
	responder := strings.TrimSpace(effect.CallbackResponderAgentID)
	if responder == "" {
		return run.RecoveryObservation{}, nil, false
	}
	if observationReader, ok := runner.options.RuntimeWorking.(RuntimeObservationReader); ok && observationReader != nil {
		return observationReader.AcquireRuntimeObservation(ctx, effect.WorkspaceID, responder)
	}
	working, known := runner.options.RuntimeWorking.RuntimeWorking(ctx, effect.WorkspaceID, responder)
	return run.RecoveryObservation{
		Working: working, Known: known,
		StateKnown: known,
		Reason:     "plain_reader",
	}, nil, known
}

// runtimeWorkingReason returns the optional diagnostic reason for a skip. It is
// empty when the reader does not implement the richer readout.
func (runner *RecoveryRunner) runtimeWorkingReason(ctx context.Context, effect run.Effect) string {
	reader, ok := runner.options.RuntimeWorking.(RuntimeWorkingReasonReader)
	if !ok || reader == nil {
		return ""
	}
	responder := strings.TrimSpace(effect.CallbackResponderAgentID)
	if responder == "" {
		return "unknown"
	}
	return reader.RuntimeWorkingReason(ctx, effect.WorkspaceID, responder)
}

func (runner *RecoveryRunner) clockNow() time.Time {
	if runner == nil || runner.options.Now == nil {
		return time.Time{}
	}
	return runner.options.Now().UTC().Truncate(time.Microsecond)
}

// sendRecoveryMessage delivers the frozen English Recovery Prompt as an
// ordinary message into the callback's frozen delivery thread
// (ThreadScope ?? DeliveryScope). It renders the resettable current-cycle
// display index separately from the effect's monotonic delivery sequence. It
// reuses the frozen MessageRequest facts from the durable effect request and
// the frozen receipt facts, never re-resolving owner, destination, or Team
// context. It keeps Callback=true so the ordinary
// message transaction routes the nudge through the same trusted responder-ref
// path that wakes the frozen responder (RequireSingleWorkflowCallbackResponder),
// and pins the frozen responder Agent so a drifted responder fails closed with
// zero side effects. The RecoveryRunner
// never persists a new awaiting effect and never re-gates execution capacity.
func (runner *RecoveryRunner) sendRecoveryMessage(ctx context.Context, effect run.Effect, displayAttemptIndex int) error {
	request, err := DecodeEffectRequest(effect.RequestJSON)
	if err != nil {
		return fmt.Errorf("decode workflow Run recovery effect request: %w", err)
	}
	if request.Message == nil {
		return fmt.Errorf("workflow Run recovery effect has no frozen message request")
	}
	receipt, err := DecodeEffectResult(effect.ResultJSON)
	if err != nil {
		return fmt.Errorf("decode workflow Run recovery effect receipt: %w", err)
	}
	targetScope := receipt.ThreadScope
	if targetScope == nil {
		targetScope = receipt.DeliveryScope
	}
	if targetScope == nil {
		return fmt.Errorf("workflow Run recovery effect has no frozen delivery scope")
	}
	frozen := *request.Message
	// The durable frozen winner is the sole authority for the recovery nudge.
	// A fresh dispatch materializes a selectOne winner into ResponderAgentRef and
	// clears the candidate pool (dispatcher.go); the durable effect request keeps
	// only the candidate pool, so a recovery resend that merely copies the request
	// would deliver with an empty responder ref and an unresolved pool. Reuse the
	// same authoritative materialization from the durable receipt/effect frozen
	// CallbackResponderAgentID: set the responder ref and drop the pool. The
	// ordinary message transaction then routes exactly to the winner and is
	// fenced by the runtime Agent coordinate below, with no re-selection
	// and no additional capacity claim.
	frozenResponder := strings.TrimSpace(effect.CallbackResponderAgentID)
	if frozenResponder == "" {
		frozenResponder = strings.TrimSpace(receipt.CallbackResponderAgentID)
	}
	if len(frozen.ResponderCandidates) > 0 {
		if frozenResponder == "" {
			return fmt.Errorf("workflow Run recovery effect has a candidate pool but no durable frozen winner")
		}
		frozen.ResponderAgentRef = frozenResponder
		frozen.ResponderCandidates = nil
	}
	// Keep Callback=true and the frozen ResponderAgentRef so the responder is
	// woken through the same trusted responder-ref chain as the original
	// delivery, with the internal fence guaranteeing the frozen ID is never
	// re-resolved to a different Agent.
	frozen.TargetRef = ""
	frozen.StartThread = false
	frozen.WorkScopeMode = ""
	frozen.WorkScope = nil
	frozen.RelatedScopes = nil
	frozen.TaskWatchHandoff = nil
	frozen.ExpectedCallbackResponderAgentID = strings.TrimSpace(receipt.CallbackResponderAgentID)
	// The recovery nudge is a separate wake message for the already-awaiting
	// callback; it must not be bound to the awaiting effect as if it were the
	// original delivery. The fanout writer skips the workflow_run contribution
	// source binding for this flag while keeping the workflow context metadata
	// and the trusted responder fence.
	frozen.SkipWorkflowRunMessageBinding = true
	scope := *targetScope
	frozen.Scope = &scope
	if effect.RecoveryAttemptCount < 1 {
		return fmt.Errorf("workflow Run recovery effect has no durable delivery sequence")
	}
	frozen.Content = renderRecoveryPrompt(effect.RunID, effect.ID, displayAttemptIndex)
	frozen.Presentation = domain.NewWorkflowRecoveryPresentation(
		effect.RunID, effect.ID, effect.ID, displayAttemptIndex, domain.WorkflowRecoveryMaxAttempts,
	)
	frozen.ClientMessageID = recoveryClientMessageID(effect.RunID, effect.ID, effect.RecoveryAttemptCount)
	if _, err := runner.messages.WriteRunMessage(ctx, frozen); err != nil {
		return err
	}
	return nil
}