/**
 * Play Execution Governor — the deep module that owns execution policy.
 *
 * The CJS runner gates work through one Governor instance per run-attempt. The
 * runner owns execution mechanics; the Governor owns the "may I, and how many
 * at once" policy. See ADR 0007 + CONTEXT.md.
 *
 * Surface (small, by design):
 *   - acquireRowSlot / acquireToolSlot / acquireIntegrationRequestSlot
 *                                                               → local execution leases
 *   - acquireProviderPermit                                   → per-attempt provider admission
 *   - chargeBudget                                            → throws on breach
 *   - forkInlineChild                                         → inline recursion scope
 *   - resolveRowConcurrency / reportProviderBackpressure / snapshot
 */
import {
  type AdapterId,
  type ResolvedExecutionPolicy,
  resolveExecutionPolicy,
  resolveMaxConcurrentExternalCalls,
  resolveRowConcurrency,
} from './policy';
import { type PacingRule, type RateStateBackend } from './rate-state-backend';
import {
  createInMemoryAdaptiveAdmission,
  type RuntimeAdaptiveAdmission,
} from './adaptive-admission';
import {
  BudgetStateLimitError,
  InMemoryBudgetStateBackend,
  type BudgetStateBackend,
} from './budget-state-backend';

export interface WorkLease {
  /** Free the slot / pacing permit. Idempotent. MUST be called in a finally. */
  release(): void;
}

export type BudgetKind = 'toolCall' | 'retry' | 'waterfallStep';

/** Runtime identity and counters shared by inline composition. */
export interface GovernanceSnapshot {
  rootRunId: string;
  currentRunId: string;
  currentPlayId: string;
  ancestryPlayIds: string[];
  ancestryRunIds: string[];
  callDepth: number;
  toolCallCount: number;
  retryCount: number;
  waterfallStepExecutions: number;
}

export class GovernorBudgetError extends Error {
  constructor(
    readonly budget: BudgetKind | 'playDepth',
    readonly observed: number,
    readonly limit: number,
  ) {
    const message =
      budget === 'playDepth'
        ? `Play-call depth exceeded (${observed}/${limit}).`
        : `Play execution ${budget} budget exceeded (${observed}/${limit}).`;
    super(message);
    this.name = 'GovernorBudgetError';
  }
}

/** Maps a toolId to its provider + resolved pacing rules (from rate-limit defs). */
export type PacingResolver = (
  toolId: string,
) => Promise<{ provider: string; rules: PacingRule[] } | null>;

export type RateScopeResolver = (
  toolId: string,
  provider: string,
) =>
  | Promise<{ bucketId: string; token: string } | null>
  | { bucketId: string; token: string }
  | null;

export function defaultPacingForTool(
  toolId: string,
  policy: ResolvedExecutionPolicy,
): { provider: string; rules: PacingRule[] } {
  return {
    // Unknown providers still need a stable bucket. Tool id is the only
    // substrate-neutral key available when the runtime has no queue hints.
    provider: `tool:${toolId}`,
    rules: [
      {
        ruleId: `default:${toolId}`,
        requestsPerWindow: policy.pacing.defaultProviderRequestsPerSecond,
        windowMs: 1_000,
        maxConcurrency: null,
      },
    ],
  };
}

export interface PlayExecutionGovernor {
  readonly adapter: AdapterId;
  readonly policy: ResolvedExecutionPolicy;

  /** Block until a map-row slot is free. */
  acquireRowSlot(opts?: { signal?: AbortSignal }): Promise<WorkLease>;
  /**
   * Acquire the local execution slot and charge one logical tool-call budget.
   * Provider pacing deliberately happens separately at the HTTP-attempt edge.
   */
  acquireToolSlot(
    toolId: string,
    opts?: { signal?: AbortSignal },
  ): Promise<WorkLease>;

  /**
   * Admit one brief runtime-to-app HTTP fetch/body exchange. Long provider or
   * fixture residence happens before this edge and must not occupy this slot.
   */
  acquireIntegrationRequestSlot(opts?: {
    signal?: AbortSignal;
  }): Promise<WorkLease>;

  /**
   * Block until one physical provider HTTP attempt is admitted under the
   * provider's shared rate/concurrency rules. Call this immediately before the
   * outbound request and again for every retry.
   */
  acquireProviderPermit(
    toolId: string,
    opts?: { signal?: AbortSignal },
  ): Promise<WorkLease>;

  /**
   * Suggested batch parallelism for a tool: the provider's own rate hints
   * tightened to the policy's suggested ceiling. No hints → the fallback.
   */
  suggestedParallelism(toolId: string, fallback: number): Promise<number>;

  /** Increment a monotonic budget counter; throws GovernorBudgetError on breach. */
  chargeBudget(kind: BudgetKind, amount?: number): Promise<void>;

  /**
   * Fork an inline child view. The child gets its own recursion lineage while
   * sharing the parent's row/tool semaphores, adaptive admission, and budgets.
   */
  forkInlineChild(input: {
    childPlayName: string;
    childRunId: string;
  }): Promise<PlayExecutionGovernor>;

  /** Effective row concurrency: explicit request clamped to [1, rowMax], else default. */
  resolveRowConcurrency(requested?: number): number;

  /** Feed a provider's Retry-After back into the shared pacer. */
  reportProviderBackpressure(input: {
    provider: string;
    /**
     * The tool whose provider call received the 429. Credential-scoped rate
     * buckets are resolved per tool, so delayed feedback must retain this
     * identity instead of selecting the latest scope for the provider.
     */
    toolId?: string;
    retryAfterMs: number;
  }): Promise<void>;

  /** Feed clean provider calls back into adaptive admission. */
  observeProviderSuccess(input: {
    provider: string;
    toolId?: string;
    latencyMs?: number | null;
  }): void;
  /** Feed success back using the same tool-to-provider resolution as acquire. */
  observeToolSuccess(input: {
    toolId: string;
    latencyMs?: number | null;
  }): void;

  snapshot(): GovernanceSnapshot;
}

interface GovernorInput {
  adapter: AdapterId;
  scope: { orgId: string; rootRunId: string };
  rateState: RateStateBackend;
  budgetState?: BudgetStateBackend;
  resolvePacing: PacingResolver;
  resolveRateScope?: RateScopeResolver;
  adaptiveAdmission?: RuntimeAdaptiveAdmission;
  resume?: GovernanceSnapshot;
  maxConcurrentExternalCalls?: number | null;
  maxConcurrentRows?: number | null;
}

class Semaphore {
  private inFlight = 0;
  private readonly waiters: Array<() => void> = [];
  constructor(private readonly limit: number) {}

  async acquire(signal?: AbortSignal): Promise<WorkLease> {
    // Fail fast on an already-aborted signal: the parked-promise abort listener
    // below registers with { once: true } and never fires for a signal that was
    // aborted before we parked, so without this check a full pool would block
    // the waiter until a slot frees (or forever if it never drains).
    if (signal?.aborted) {
      throw signal.reason instanceof Error
        ? signal.reason
        : new Error('Slot acquire aborted.');
    }
    while (this.inFlight >= this.limit) {
      await new Promise<void>((resolve, reject) => {
        const onResolve = () => {
          signal?.removeEventListener('abort', onAbort);
          resolve();
        };
        const onAbort = () => {
          const idx = this.waiters.indexOf(onResolve);
          if (idx >= 0) this.waiters.splice(idx, 1);
          reject(
            signal?.reason instanceof Error
              ? signal.reason
              : new Error('Slot acquire aborted.'),
          );
        };
        this.waiters.push(onResolve);
        signal?.addEventListener('abort', onAbort, { once: true });
      });
      if (signal?.aborted) {
        throw signal.reason instanceof Error
          ? signal.reason
          : new Error('Slot acquire aborted.');
      }
    }
    this.inFlight += 1;
    let released = false;
    return {
      release: () => {
        if (released) return;
        released = true;
        this.inFlight = Math.max(0, this.inFlight - 1);
        this.waiters.shift()?.();
      },
    };
  }
}

export function createDefaultGovernanceSnapshot(scope: {
  orgId: string;
  rootRunId: string;
  rootPlayId?: string;
}): GovernanceSnapshot {
  return {
    rootRunId: scope.rootRunId,
    currentRunId: scope.rootRunId,
    currentPlayId: scope.rootPlayId ?? scope.rootRunId,
    ancestryPlayIds: scope.rootPlayId ? [scope.rootPlayId] : [],
    ancestryRunIds: [scope.rootRunId],
    callDepth: scope.rootPlayId ? 1 : 0,
    toolCallCount: 0,
    retryCount: 0,
    waterfallStepExecutions: 0,
  };
}

export function createPlayExecutionGovernor(
  input: GovernorInput,
): PlayExecutionGovernor {
  const basePolicy = resolveExecutionPolicy(input.adapter);
  const maxConcurrentExternalCalls = resolveMaxConcurrentExternalCalls(
    input.maxConcurrentExternalCalls,
  );
  const maxInFlightDispatchGroups = Math.min(
    maxConcurrentExternalCalls,
    basePolicy.concurrency.toolDispatchGroups,
  );
  const policy: ResolvedExecutionPolicy = {
    ...basePolicy,
    concurrency: {
      ...basePolicy.concurrency,
      toolCalls: maxConcurrentExternalCalls,
      toolDispatchGroups: maxInFlightDispatchGroups,
      toolDispatchGroupsPerLane: maxInFlightDispatchGroups,
    },
  };
  const state: GovernanceSnapshot =
    input.resume ?? createDefaultGovernanceSnapshot(input.scope);

  const rowSlots = new Semaphore(policy.concurrency.rowMax);
  const toolSlots = new Semaphore(policy.concurrency.toolCalls);
  const integrationRequestSlots = new Semaphore(
    policy.concurrency.integrationRequests,
  );
  const adaptiveAdmission =
    input.adaptiveAdmission ??
    createInMemoryAdaptiveAdmission({
      initialRequestsPerSecond: policy.pacing.defaultProviderRequestsPerSecond,
      maxRequestsPerSecond: policy.pacing.suggestedMaxParallelism,
    });
  const budgetState = input.budgetState ?? new InMemoryBudgetStateBackend();
  const providerByTool = new Map<string, string>();
  const pacingRulesByTool = new Map<string, PacingRule[]>();
  const scopeByTool = new Map<string, [string, string]>();
  const providerConcurrencySlotsByTool = new Map<string, Semaphore>();

  // When the rate-state backend owns pacing authoritatively (the Absurd/Node
  // app-runtime Postgres pacer runs the whole token bucket + AIMD in the row),
  // the Governor's in-memory adaptive admission becomes a passthrough: it only
  // SEEDS declared/default rules and never halves/ramps them here. Backpressure
  // is routed solely to `rateState.penalize` so the DB row is the single source
  // of pacing truth. The coordinator and in-memory backends leave `kind`
  // undefined and keep the in-memory adaptive admission (byte-identical).
  const dbAuthoritativePacing = input.rateState.kind === 'app_runtime_postgres';

  const orgBucket = (provider: string) => `${input.scope.orgId}:${provider}`;

  async function resolveScope(toolId: string, provider: string) {
    const resolved = await input.resolveRateScope?.(toolId, provider);
    const scope = [
      resolved?.bucketId ?? orgBucket(provider),
      resolved?.token ?? '',
    ] as [string, string];
    scopeByTool.set(toolId, scope);
    return scope;
  }

  const scopeFor = (toolId: string | undefined, provider: string) =>
    scopeByTool.get(toolId ?? '') ??
    ([orgBucket(provider), ''] as [string, string]);

  async function resolveAdaptivePacing(toolId: string): Promise<{
    provider: string;
    rules: PacingRule[];
  }> {
    const resolvedPacing = await input.resolvePacing(toolId);
    const declaredPacing =
      resolvedPacing && resolvedPacing.rules.length > 0 ? resolvedPacing : null;
    if (dbAuthoritativePacing) {
      // Passthrough: hand the DB-backed pacer the declared rules verbatim (or
      // the default-pacing shape when undeclared). No in-memory admission — the
      // row is authoritative.
      const pacing = declaredPacing ?? defaultPacingForTool(toolId, policy);
      providerByTool.set(toolId, pacing.provider);
      pacingRulesByTool.set(toolId, pacing.rules);
      return pacing;
    }
    const pacing = adaptiveAdmission.resolvePacing({
      orgId: input.scope.orgId,
      toolId,
      declaredPacing,
      fallbackPacing: defaultPacingForTool(toolId, policy),
    });
    providerByTool.set(toolId, pacing.provider);
    pacingRulesByTool.set(toolId, pacing.rules);
    return pacing;
  }

  const budgetLimit = (kind: BudgetKind): number => {
    switch (kind) {
      case 'toolCall':
        return policy.budgets.maxToolCallCount;
      case 'retry':
        return policy.budgets.maxRetryCount;
      case 'waterfallStep':
        return policy.budgets.maxWaterfallStepExecutions;
    }
  };

  async function reserveBudgets(
    charges: Array<{ kind: BudgetKind; amount: number }>,
  ): Promise<void> {
    try {
      await budgetState.charge({
        rootRunId: state.rootRunId,
        charges: charges.map((charge) => ({
          key: charge.kind,
          amount: charge.amount,
          limit: budgetLimit(charge.kind),
        })),
      });
    } catch (error) {
      if (error instanceof BudgetStateLimitError) {
        throw new GovernorBudgetError(
          error.key as BudgetKind,
          error.observed,
          error.limit,
        );
      }
      throw error;
    }
  }

  async function chargeBudget(kind: BudgetKind, amount = 1): Promise<void> {
    const current =
      kind === 'toolCall'
        ? state.toolCallCount
        : kind === 'retry'
          ? state.retryCount
          : state.waterfallStepExecutions;
    if (current + amount > budgetLimit(kind)) {
      throw new GovernorBudgetError(kind, current + amount, budgetLimit(kind));
    }
    await reserveBudgets([{ kind, amount }]);
    switch (kind) {
      case 'toolCall':
        state.toolCallCount += amount;
        return;
      case 'retry':
        state.retryCount += amount;
        return;
      case 'waterfallStep':
        state.waterfallStepExecutions += amount;
        return;
    }
  }

  const governor: PlayExecutionGovernor = {
    adapter: input.adapter,
    policy,

    acquireRowSlot: (opts) => rowSlots.acquire(opts?.signal),
    async acquireToolSlot(_toolId, opts) {
      const slot = await toolSlots.acquire(opts?.signal);
      // Charge the logical call only after its local execution slot is held, so
      // a failed/aborted slot acquisition never consumes the budget. The
      // physical provider admission is intentionally deferred to
      // acquireProviderPermit at the outbound HTTP edge.
      try {
        await chargeBudget('toolCall');
      } catch (error) {
        slot.release();
        throw error;
      }
      return slot;
    },

    acquireIntegrationRequestSlot: (opts) =>
      integrationRequestSlots.acquire(opts?.signal),

    async acquireProviderPermit(toolId, opts) {
      // The rate ticket is an admission for a physical outbound attempt, not a
      // reservation made while receipt ownership, headers, or local tool slots
      // may still be waiting. This is what keeps real provider arrivals inside
      // the declared rolling window under concurrent runners.
      const pacing = await resolveAdaptivePacing(toolId);
      const rateScope = await resolveScope(toolId, pacing.provider);
      const declaredMaxConcurrency = Math.min(
        ...pacing.rules.flatMap((rule) =>
          rule.maxConcurrency != null ? [rule.maxConcurrency] : [],
        ),
      );
      const localConcurrencySlot = Number.isFinite(declaredMaxConcurrency)
        ? await (() => {
            let slots = providerConcurrencySlotsByTool.get(toolId);
            if (!slots) {
              slots = new Semaphore(Math.max(1, declaredMaxConcurrency));
              providerConcurrencySlotsByTool.set(toolId, slots);
            }
            return slots.acquire(opts?.signal);
          })()
        : null;
      try {
        const rateLease = await input.rateState.acquire({
          bucketId: rateScope[0],
          rateScopeToken: rateScope[1],
          rules: pacing.rules,
          signal: opts?.signal,
        });
        return {
          release() {
            rateLease.release();
            localConcurrencySlot?.release();
          },
        };
      } catch (error) {
        localConcurrencySlot?.release();
        throw error;
      }
    },

    async suggestedParallelism(toolId, fallback) {
      const pacing = await resolveAdaptivePacing(toolId);
      const limits = pacing.rules.flatMap((rule) =>
        rule.maxConcurrency != null
          ? [rule.requestsPerWindow, rule.maxConcurrency]
          : [rule.requestsPerWindow],
      );
      return Math.max(
        1,
        Math.min(policy.pacing.suggestedMaxParallelism, ...limits),
      );
    },

    chargeBudget,

    async forkInlineChild(childInput) {
      return createInlineChildGovernor(
        governor,
        deriveInlineChildSnapshot(state, childInput, policy),
      );
    },

    resolveRowConcurrency: (requested) =>
      resolveRowConcurrency(policy, requested, input.maxConcurrentRows),

    async reportProviderBackpressure(bp) {
      // DB-authoritative pacer: route backpressure ONLY to the row via
      // penalize. The in-memory adaptive admission must not also halve, or the
      // provider would be throttled twice (once in the row, once in memory).
      if (!dbAuthoritativePacing) {
        adaptiveAdmission.observeProviderBackpressure({
          orgId: input.scope.orgId,
          provider: bp.provider,
          retryAfterMs: bp.retryAfterMs,
        });
      }
      const rateScope = scopeFor(bp.toolId, bp.provider);
      await input.rateState.penalize({
        bucketId: rateScope[0],
        ...(rateScope[1] ? { rateScopeToken: rateScope[1] } : {}),
        cooldownMs: bp.retryAfterMs,
      });
    },

    observeProviderSuccess(success) {
      if (dbAuthoritativePacing) {
        if (!success.toolId) return;
        const rateScope = scopeFor(success.toolId, success.provider);
        const rules = pacingRulesByTool.get(success.toolId);
        if (!rules) return;
        input.rateState.observeSuccess?.({
          bucketId: rateScope[0],
          rules,
        });
        return;
      }
      adaptiveAdmission.observeProviderSuccess({
        orgId: input.scope.orgId,
        provider: success.provider,
        latencyMs: success.latencyMs,
      });
    },

    observeToolSuccess(success) {
      const provider =
        providerByTool.get(success.toolId) ??
        defaultPacingForTool(success.toolId, policy).provider;
      governor.observeProviderSuccess({
        provider,
        toolId: success.toolId,
        latencyMs: success.latencyMs,
      });
    },

    snapshot: () => ({
      ...state,
      ancestryPlayIds: [...state.ancestryPlayIds],
      ancestryRunIds: [...state.ancestryRunIds],
    }),
  };
  return governor;
}

function deriveInlineChildSnapshot(
  parent: GovernanceSnapshot,
  input: { childPlayName: string; childRunId: string },
  policy: ResolvedExecutionPolicy,
): GovernanceSnapshot {
  if (parent.ancestryPlayIds.includes(input.childPlayName)) {
    throw new Error(
      `Recursive play graph detected: ${[
        ...parent.ancestryPlayIds,
        input.childPlayName,
      ].join(' -> ')}`,
    );
  }
  const nextDepth = parent.callDepth + 1;
  if (nextDepth > policy.budgets.maxPlayCallDepth) {
    throw new GovernorBudgetError(
      'playDepth',
      nextDepth,
      policy.budgets.maxPlayCallDepth,
    );
  }
  return {
    ...parent,
    currentRunId: input.childRunId,
    currentPlayId: input.childPlayName,
    ancestryPlayIds: [...parent.ancestryPlayIds, input.childPlayName],
    ancestryRunIds: [...parent.ancestryRunIds, input.childRunId],
    callDepth: nextDepth,
  };
}

function createInlineChildGovernor(
  root: PlayExecutionGovernor,
  initial: GovernanceSnapshot,
): PlayExecutionGovernor {
  const lineage = {
    rootRunId: initial.rootRunId,
    currentRunId: initial.currentRunId,
    currentPlayId: initial.currentPlayId,
    ancestryPlayIds: [...initial.ancestryPlayIds],
    ancestryRunIds: [...initial.ancestryRunIds],
    callDepth: initial.callDepth,
  };

  const child: PlayExecutionGovernor = {
    adapter: root.adapter,
    policy: root.policy,
    acquireRowSlot: (opts) => root.acquireRowSlot(opts),
    acquireToolSlot: (toolId, opts) => root.acquireToolSlot(toolId, opts),
    acquireIntegrationRequestSlot: (opts) =>
      root.acquireIntegrationRequestSlot(opts),
    acquireProviderPermit: (toolId, opts) =>
      root.acquireProviderPermit(toolId, opts),
    suggestedParallelism: (toolId, fallback) =>
      root.suggestedParallelism(toolId, fallback),
    chargeBudget: (kind, amount) => root.chargeBudget(kind, amount),
    resolveRowConcurrency: (requested) => root.resolveRowConcurrency(requested),
    reportProviderBackpressure: async (input) =>
      await root.reportProviderBackpressure(input),
    observeProviderSuccess: (input) => root.observeProviderSuccess(input),
    observeToolSuccess: (input) => root.observeToolSuccess(input),
    async forkInlineChild(input) {
      return createInlineChildGovernor(
        child,
        deriveInlineChildSnapshot(child.snapshot(), input, child.policy),
      );
    },
    snapshot() {
      const counters = root.snapshot();
      return {
        ...counters,
        ...lineage,
        ancestryPlayIds: [...lineage.ancestryPlayIds],
        ancestryRunIds: [...lineage.ancestryRunIds],
      };
    },
  };
  return child;
}