import { normalizeTableNamespace } from './row-identity';
import { sqlSafePlayColumnName } from '../play-data-plane/column-names';
import type { PlayDocflow } from './docflow';

export { sqlSafePlayColumnName };

/**
 * A top-level key the play's function literally `return`s. Derived from the
 * `return { ... }` object literal — NOT from dataset `.withColumn(...)` names —
 * so the "Returns" graph node mirrors the function's real output shape.
 * `isDataset` is true when the key's value is a `PlayDataset` handle (a table).
 */
export interface PlayStaticReturnField {
  name: string;
  isDataset: boolean;
  /**
   * For a dataset-valued field, the `ctx.dataset(KEY, ...)` key backing it
   * (e.g. `job_change_checks`) — so the UI can label the return by the dataset
   * it actually produces instead of the bland object key (`rows`). Undefined
   * for non-dataset fields, `ctx.csv(...)` (no durable name), or when the key
   * isn't a static string literal.
   */
  datasetName?: string;
}

export interface PlayStaticPipeline {
  /** Authored business flow. Static analysis is used only when this is absent. */
  docflow?: PlayDocflow;
  tableNamespace?: string;
  inputFields?: string[];
  rowKeyFields?: string[];
  csvArg?: string;
  hasInlineData?: boolean;
  csvDescription?: string;
  datasetDescription?: string;
  fields: string[];
  /**
   * Top-level keys of the play's `return { ... }` object literal, in source
   * order. Undefined when the terminal return isn't a statically-known object
   * literal (bare value, dataset handle, conditional returns, etc.).
   */
  returnFields?: PlayStaticReturnField[];
  stages?: PlayStaticSubstep[];
  substeps: PlayStaticSubstep[];
  sheetContract?: PlaySheetContract | null;
  sheetContractErrors?: string[];
}

/**
 * Durable V2 representation for static pipelines. Convex documents have a
 * finite nesting limit, so this stores the recursive graph as flat records.
 * The V1 `staticPipeline` field remains a small compatibility projection for
 * legacy readers; V2 is rehydrated before current readers consume it.
 */
export interface StoredStaticPipelineV2 {
  version: 2;
  rootPipelineId: string;
  pipelines: StoredStaticPipelineV2Pipeline[];
  nodes: StoredStaticPipelineV2Node[];
  branches: StoredStaticPipelineV2Branch[];
}

export interface StoredStaticPipelineV2Pipeline {
  id: string;
  parentNodeId?: string;
  value: Record<string, unknown>;
}

export interface StoredStaticPipelineV2Node {
  id: string;
  pipelineId: string;
  parentNodeId?: string;
  branchId?: string;
  container: 'stages' | 'substeps' | 'steps' | 'branch_steps';
  index: number;
  nestedPipelineId?: string;
  value: Record<string, unknown>;
}

export interface StoredStaticPipelineV2Branch {
  id: string;
  nodeId: string;
  index: number;
  label: string;
  condition?: string;
}

function dedupeStaticFieldNames(
  fields: Array<string | null | undefined>,
): string[] {
  const seen = new Set<string>();
  const out: string[] = [];
  for (const field of fields) {
    const trimmed = typeof field === 'string' ? field.trim() : '';
    if (!trimmed || seen.has(trimmed)) {
      continue;
    }
    seen.add(trimmed);
    out.push(trimmed);
  }
  return out;
}

export function inputFieldFromStaticCsvArg(csvArg: unknown): string | null {
  if (typeof csvArg !== 'string') return null;
  const match = /^input\.([A-Za-z_$][\w$]*)$/.exec(csvArg.trim());
  return match?.[1] ?? null;
}

export function deriveStaticPipelineFileInputFields(
  pipeline: PlayStaticPipeline | null | undefined,
): string[] {
  if (!pipeline) {
    return [];
  }
  const explicitCsvFields = getCompiledPipelineSubsteps(pipeline)
    .filter((substep) => substep.type === 'csv')
    .map((substep) =>
      substep.type === 'csv'
        ? (inputFieldFromStaticCsvArg(substep.path) ?? substep.field)
        : null,
    );
  const inferredCsvField =
    explicitCsvFields.length > 0
      ? null
      : (inputFieldFromStaticCsvArg(pipeline.csvArg) ??
        (pipeline.csvArg ? 'csv' : null));
  return dedupeStaticFieldNames([...explicitCsvFields, inferredCsvField]);
}

export function deriveStaticPipelineEntryInputFields(
  pipeline: PlayStaticPipeline | null | undefined,
): string[] {
  if (!pipeline) {
    return [];
  }
  const fileInputs = deriveStaticPipelineFileInputFields(pipeline);
  return fileInputs.length > 0
    ? fileInputs
    : dedupeStaticFieldNames(pipeline.inputFields ?? []);
}

export type PlaySheetColumnSource =
  | 'input'
  | 'datasetColumn'
  | 'waterfallStep'
  | 'childPlayColumn';

export interface PlaySheetColumnContract {
  id: string;
  sqlName: string;
  source: PlaySheetColumnSource;
  field?: string;
  parentField?: string;
  playId?: string;
  waterfallId?: string;
  outputField?: string;
  outputSqlName?: string;
  stepId?: string;
  toolId?: string;
  isRowKey?: boolean;
}

export interface PlaySheetContract {
  tableNamespace: string;
  columns: PlaySheetColumnContract[];
}

export type PlayStaticColumnProducerKind =
  | 'tool'
  | 'waterfall'
  | 'stepProgram'
  | 'playCall'
  | 'controlFlow'
  | 'transform';

export interface PlayStaticColumnProducer {
  id: string;
  kind: PlayStaticColumnProducerKind;
  field: string;
  toolId?: string;
  playId?: string;
  conditional?: boolean;
  sourceRange?: PlayStaticSourceRange;
  steps?: PlayStaticColumnProducer[];
  substep: PlayStaticSubstep;
}

export interface PlayStaticDatasetColumn {
  id: string;
  source: PlaySheetColumnSource;
  sqlName?: string;
  producers: PlayStaticColumnProducer[];
}

export interface PlayCompiledStaticGraph {
  topLevel: PlayStaticSubstep[];
  datasets: Array<{
    tableNamespace: string;
    columns: PlayStaticDatasetColumn[];
  }>;
}

export function ensureCompiledSheetContract(
  pipeline: PlayStaticPipeline | null | undefined,
): PlayStaticPipeline | null | undefined {
  if (!pipeline) {
    return pipeline;
  }
  const datasetErrors: string[] = [];
  const compileDatasetSubsteps = (
    substeps: PlayStaticSubstep[],
  ): PlayStaticSubstep[] =>
    substeps.map((substep) => {
      if (substep.type === 'play_call' && substep.pipeline) {
        return {
          ...substep,
          pipeline: ensureCompiledSheetContract(substep.pipeline) ?? null,
        };
      }
      if (
        substep.type !== 'dataset' &&
        substep.type !== 'step_suite' &&
        substep.type !== 'control_flow'
      ) {
        return substep;
      }
      const steps = substep.steps?.length
        ? compileDatasetSubsteps(substep.steps)
        : (substep.steps ?? []);
      if (substep.type !== 'dataset' || substep.sheetContract) {
        return { ...substep, steps } as PlayStaticSubstep;
      }
      const tableNamespace = (substep.tableNamespace ?? substep.field).trim();
      const compiledDataset = compileSheetContract({
        tableNamespace,
        inputFields: substep.inputFields,
        rowKeyFields: substep.rowKeyFields,
        fields:
          substep.outputFields ??
          substep.columns?.map((column) => column.id) ??
          [],
        substeps: steps,
      });
      datasetErrors.push(
        ...compiledDataset.errors.map(
          (error) => `Dataset "${tableNamespace}": ${error}`,
        ),
      );
      return {
        ...substep,
        steps,
        sheetContract: compiledDataset.contract,
      };
    });
  const withDatasetContracts: PlayStaticPipeline = {
    ...pipeline,
    ...(pipeline.stages
      ? { stages: compileDatasetSubsteps(pipeline.stages) }
      : {}),
    // Stored legacy rows can predate the required static-pipeline arrays.
    // Preserve the missing field so compileSheetContract reports it instead
    // of crashing or silently normalizing malformed persisted data.
    ...(Array.isArray(pipeline.substeps)
      ? { substeps: compileDatasetSubsteps(pipeline.substeps) }
      : {}),
  };
  const compiled = pipeline.sheetContract
    ? {
        contract: pipeline.sheetContract,
        errors: [...(pipeline.sheetContractErrors ?? [])],
      }
    : compileSheetContract(withDatasetContracts);
  return {
    ...withDatasetContracts,
    sheetContract: compiled.contract,
    sheetContractErrors: [...compiled.errors, ...datasetErrors],
  };
}

const DEFAULT_MAX_EMBEDDED_PLAY_CALL_PIPELINE_DEPTH = 3;
const DEFAULT_MAX_STORED_SUBSTEP_DEPTH = 3;

function cloneStorageSafeSourceRange(
  sourceRange: PlayStaticSourceRange | undefined,
): PlayStaticSourceRange | undefined {
  return sourceRange ? { ...sourceRange } : undefined;
}

function cloneStorageSafeSheetContract(
  contract: PlaySheetContract | null | undefined,
): PlaySheetContract | null | undefined {
  if (!contract) return contract;
  return {
    ...contract,
    columns: contract.columns.map((column) => ({ ...column })),
  };
}

function omitUndefinedProperties<T extends Record<string, unknown>>(
  value: T,
): T {
  const out: Record<string, unknown> = {};
  for (const [key, entry] of Object.entries(value)) {
    if (entry !== undefined) {
      out[key] = entry;
    }
  }
  return out as T;
}

function stripUndefinedDeep<T>(value: T): T {
  if (Array.isArray(value)) {
    return value.map((entry) => stripUndefinedDeep(entry)) as T;
  }
  if (value && typeof value === 'object') {
    const out: Record<string, unknown> = {};
    for (const [key, entry] of Object.entries(value)) {
      if (entry !== undefined) {
        out[key] = stripUndefinedDeep(entry);
      }
    }
    return out as T;
  }
  return value;
}

function truncateStaticControlFlowBranchesForStorage(
  branches: PlayStaticControlFlowBranch[] | undefined,
  input: {
    embeddedPlayCallPipelineDepth: number;
    maxEmbeddedPlayCallPipelineDepth: number;
    maxStoredSubstepDepth: number;
    storedSubstepDepth: number;
  },
): PlayStaticControlFlowBranch[] | undefined {
  if (!branches) return undefined;
  return branches.map((branch) =>
    omitUndefinedProperties({
      label: branch.label,
      condition: branch.condition,
      steps: truncateStaticSubstepsForStorage(
        branch.steps,
        nextStoredSubstepInput(input),
      ),
    }),
  );
}

function nextStoredSubstepInput(input: {
  embeddedPlayCallPipelineDepth: number;
  maxEmbeddedPlayCallPipelineDepth: number;
  maxStoredSubstepDepth: number;
  storedSubstepDepth: number;
}) {
  return {
    ...input,
    storedSubstepDepth: input.storedSubstepDepth + 1,
  };
}

function truncateStaticSubstepShallowForStorage(
  substep: Record<string, unknown>,
): PlayStaticSubstep {
  const shallow: Record<string, unknown> = { ...substep };
  delete shallow.branches;
  delete shallow.pipeline;
  delete shallow.steps;

  if (
    shallow.type === 'control_flow' ||
    shallow.type === 'dataset' ||
    shallow.type === 'step_suite'
  ) {
    shallow.steps = [];
  }

  return omitUndefinedProperties(shallow) as PlayStaticSubstep;
}

function truncateStaticSubstepsForStorage(
  substeps: PlayStaticSubstep[] | undefined,
  input: {
    embeddedPlayCallPipelineDepth: number;
    maxEmbeddedPlayCallPipelineDepth: number;
    maxStoredSubstepDepth: number;
    storedSubstepDepth: number;
  },
): PlayStaticSubstep[] {
  return (substeps ?? []).map((substep) => {
    const {
      paramsSource: _paramsSource,
      returnSource: _returnSource,
      sourceText: _sourceText,
      ...substepWithoutSourceText
    } = substep as PlayStaticSubstep & {
      paramsSource?: string;
      pipeline?: PlayStaticPipeline | null;
      returnSource?: string;
      sourceText?: string;
    };
    const base = omitUndefinedProperties({
      ...substepWithoutSourceText,
      sourceRange: cloneStorageSafeSourceRange(substep.sourceRange),
      callPath: substep.callPath ? [...substep.callPath] : undefined,
    });

    if (input.storedSubstepDepth >= input.maxStoredSubstepDepth) {
      return truncateStaticSubstepShallowForStorage(base);
    }

    if (base.type !== 'play_call' && 'pipeline' in substepWithoutSourceText) {
      const nestedPipeline = substepWithoutSourceText.pipeline;
      (base as { pipeline?: PlayStaticPipeline | null }).pipeline =
        nestedPipeline
          ? truncateStaticPipelineForStorage(nestedPipeline, {
              maxEmbeddedPlayCallPipelineDepth:
                input.maxEmbeddedPlayCallPipelineDepth,
              embeddedPlayCallPipelineDepth:
                input.embeddedPlayCallPipelineDepth + 1,
              maxStoredSubstepDepth: input.maxStoredSubstepDepth,
              storedSubstepDepth: input.storedSubstepDepth + 1,
            })
          : nestedPipeline;
    }

    const nestedInput = nextStoredSubstepInput(input);

    if (base.type === 'dataset') {
      return omitUndefinedProperties({
        ...base,
        inputFields: base.inputFields ? [...base.inputFields] : undefined,
        rowKeyFields: base.rowKeyFields ? [...base.rowKeyFields] : undefined,
        outputFields: base.outputFields ? [...base.outputFields] : undefined,
        columns: base.columns
          ? base.columns.map((column) => ({
              ...column,
              producers: column.producers.map((producer) => ({
                ...omitUndefinedProperties({ ...producer }),
                sourceRange: cloneStorageSafeSourceRange(producer.sourceRange),
                steps: producer.steps
                  ? producer.steps.map((stepProducer) => ({
                      ...omitUndefinedProperties({ ...stepProducer }),
                      sourceRange: cloneStorageSafeSourceRange(
                        stepProducer.sourceRange,
                      ),
                    }))
                  : undefined,
              })),
            }))
          : undefined,
        waterfallIds: base.waterfallIds ? [...base.waterfallIds] : undefined,
        undrawnColumns: base.undrawnColumns
          ? [...base.undrawnColumns]
          : undefined,
        steps: truncateStaticSubstepsForStorage(base.steps, nestedInput),
        sheetContract: cloneStorageSafeSheetContract(base.sheetContract),
      });
    }

    if (base.type === 'waterfall') {
      return omitUndefinedProperties({
        ...base,
        steps: base.steps?.map((step) =>
          omitUndefinedProperties({
            id: step.id,
            kind: step.kind,
            toolId: step.toolId,
          }),
        ),
      });
    }

    if (base.type === 'control_flow') {
      return omitUndefinedProperties({
        ...base,
        steps: truncateStaticSubstepsForStorage(base.steps, nestedInput),
        branches: truncateStaticControlFlowBranchesForStorage(
          base.branches,
          input,
        ),
      });
    }

    if (base.type === 'step_suite') {
      return omitUndefinedProperties({
        ...base,
        steps: truncateStaticSubstepsForStorage(base.steps, nestedInput),
      });
    }

    if (base.type !== 'play_call') {
      if (Array.isArray((base as { steps?: unknown }).steps)) {
        return omitUndefinedProperties({
          ...base,
          steps: truncateStaticSubstepsForStorage(
            (base as { steps?: PlayStaticSubstep[] }).steps,
            nestedInput,
          ),
        });
      }
      return base;
    }

    // A normalized authoring graph intentionally leaves a named play-call's
    // child pipeline absent. That is a reference edge, not a failed or
    // truncated resolution, so preserve it without inventing an error.
    if (
      !base.pipeline &&
      !Object.prototype.hasOwnProperty.call(substep, 'pipeline') &&
      !base.resolutionError
    ) {
      return base;
    }

    if (
      !base.pipeline ||
      input.embeddedPlayCallPipelineDepth >=
        input.maxEmbeddedPlayCallPipelineDepth
    ) {
      return omitUndefinedProperties({
        ...base,
        pipeline: null,
        resolutionError:
          base.resolutionError ??
          `Stored static pipeline truncated at ${base.playId}`,
      });
    }

    return omitUndefinedProperties({
      ...base,
      pipeline: truncateStaticPipelineForStorage(base.pipeline, {
        maxEmbeddedPlayCallPipelineDepth:
          input.maxEmbeddedPlayCallPipelineDepth,
        embeddedPlayCallPipelineDepth: input.embeddedPlayCallPipelineDepth + 1,
        maxStoredSubstepDepth: input.maxStoredSubstepDepth,
        storedSubstepDepth: input.storedSubstepDepth + 1,
      }),
    });
  });
}

export function truncateStaticPipelineForStorage(
  pipeline: PlayStaticPipeline | null | undefined,
  options: {
    maxEmbeddedPlayCallPipelineDepth?: number;
    embeddedPlayCallPipelineDepth?: number;
    maxStoredSubstepDepth?: number;
    storedSubstepDepth?: number;
  } = {},
): PlayStaticPipeline | null | undefined {
  if (!pipeline) {
    return pipeline;
  }
  const maxEmbeddedPlayCallPipelineDepth =
    options.maxEmbeddedPlayCallPipelineDepth ??
    DEFAULT_MAX_EMBEDDED_PLAY_CALL_PIPELINE_DEPTH;
  const embeddedPlayCallPipelineDepth =
    options.embeddedPlayCallPipelineDepth ?? 0;
  const maxStoredSubstepDepth =
    options.maxStoredSubstepDepth ?? DEFAULT_MAX_STORED_SUBSTEP_DEPTH;
  const storedSubstepDepth = options.storedSubstepDepth ?? 0;

  return stripUndefinedDeep(
    omitUndefinedProperties({
      ...pipeline,
      inputFields: pipeline.inputFields ? [...pipeline.inputFields] : undefined,
      rowKeyFields: pipeline.rowKeyFields
        ? [...pipeline.rowKeyFields]
        : undefined,
      fields: [...(pipeline.fields ?? [])],
      returnFields: pipeline.returnFields
        ? pipeline.returnFields.map((field) => ({ ...field }))
        : undefined,
      stages: truncateStaticSubstepsForStorage(pipeline.stages, {
        embeddedPlayCallPipelineDepth,
        maxEmbeddedPlayCallPipelineDepth,
        maxStoredSubstepDepth,
        storedSubstepDepth,
      }),
      substeps: truncateStaticSubstepsForStorage(pipeline.substeps, {
        embeddedPlayCallPipelineDepth,
        maxEmbeddedPlayCallPipelineDepth,
        maxStoredSubstepDepth,
        storedSubstepDepth,
      }),
      sheetContract: cloneStorageSafeSheetContract(pipeline.sheetContract),
      sheetContractErrors: pipeline.sheetContractErrors
        ? [...pipeline.sheetContractErrors]
        : undefined,
    }),
  );
}

export function truncateStaticPipelineForRuntimeContract(
  pipeline: PlayStaticPipeline | null | undefined,
): PlayStaticPipeline | null | undefined {
  return truncateStaticPipelineForStorage(pipeline, {
    // Runtime admission must preserve every dataset contract. The caller's
    // serialized artifact-size budget is the bound; a UI depth cap is not.
    maxStoredSubstepDepth: Number.POSITIVE_INFINITY,
  });
}

/**
 * Creates the normalized graph used at transport and control-plane seams.
 * Named child pipelines are derived from the live child definition, so retain
 * each call site but remove the recursively embedded child snapshot.
 */
export function createStaticPipelineReferenceProjection(
  pipeline: PlayStaticPipeline | null | undefined,
): PlayStaticPipeline | null | undefined {
  const stored = truncateStaticPipelineForStorage(pipeline, {
    maxEmbeddedPlayCallPipelineDepth: Number.POSITIVE_INFINITY,
    maxStoredSubstepDepth: Number.POSITIVE_INFINITY,
  });
  if (!stored) return stored;

  const projectSteps = (
    steps: PlayStaticSubstep[] | undefined,
  ): PlayStaticSubstep[] =>
    (steps ?? []).map((substep) => {
      const projected = { ...substep } as PlayStaticSubstep & {
        pipeline?: PlayStaticPipeline | null;
        branches?: Array<{
          label: string;
          condition?: string;
          steps: PlayStaticSubstep[];
        }>;
      };
      if (projected.type === 'play_call') {
        delete projected.pipeline;
        if (
          projected.resolutionError?.startsWith(
            'Stored static pipeline truncated at ',
          )
        ) {
          delete projected.resolutionError;
        }
        return projected;
      }
      if (
        (projected.type === 'dataset' ||
          projected.type === 'step_suite' ||
          projected.type === 'control_flow') &&
        Array.isArray(projected.steps)
      ) {
        projected.steps = projectSteps(projected.steps);
      }
      if (Array.isArray(projected.branches)) {
        projected.branches = projected.branches.map((branch) => ({
          ...branch,
          steps: projectSteps(branch.steps),
        }));
      }
      return projected;
    });

  return {
    ...stored,
    stages: projectSteps(stored.stages),
    substeps: projectSteps(stored.substeps),
  };
}

function asStaticRecord(value: unknown): Record<string, unknown> | null {
  return value && typeof value === 'object' && !Array.isArray(value)
    ? (value as Record<string, unknown>)
    : null;
}

function storageSafeNodeValue(
  value: Record<string, unknown>,
): Record<string, unknown> {
  const {
    branches: _branches,
    pipeline: _pipeline,
    steps: _steps,
    ...rest
  } = value;
  const out: Record<string, unknown> = { ...rest };

  // `columns[].producers[].substep` can itself contain a recursive static
  // graph. The sheet contract is the durable interface for columns, while this
  // compact metadata keeps their labels available to older graph views.
  if (Array.isArray(value.columns)) {
    out.columns = value.columns.flatMap((column) => {
      const record = asStaticRecord(column);
      if (!record) return [];
      return [
        omitUndefinedProperties({
          id: record.id,
          source: record.source,
          sqlName: record.sqlName,
        }),
      ];
    });
  }
  if (Array.isArray(value.callPath)) out.callPath = [...value.callPath];
  if (Array.isArray(value.inputFields))
    out.inputFields = [...value.inputFields];
  if (Array.isArray(value.rowKeyFields))
    out.rowKeyFields = [...value.rowKeyFields];
  if (Array.isArray(value.outputFields))
    out.outputFields = [...value.outputFields];
  if (Array.isArray(value.waterfallIds))
    out.waterfallIds = [...value.waterfallIds];
  if (Array.isArray(value.steps) && value.type === 'waterfall') {
    out.steps = value.steps.flatMap((step) => {
      const record = asStaticRecord(step);
      if (!record) return [];
      return [
        omitUndefinedProperties({
          id: record.id,
          kind: record.kind,
          toolId: record.toolId,
        }),
      ];
    });
  }
  if (asStaticRecord(value.sourceRange)) {
    out.sourceRange = { ...(value.sourceRange as Record<string, unknown>) };
  }
  if (asStaticRecord(value.sheetContract)) {
    out.sheetContract = cloneStorageSafeSheetContract(
      value.sheetContract as PlaySheetContract,
    );
  }
  return stripUndefinedDeep(out);
}

function storageSafePipelineValue(
  pipeline: PlayStaticPipeline,
): Record<string, unknown> {
  const { stages: _stages, substeps: _substeps, ...value } = pipeline;
  return stripUndefinedDeep(
    omitUndefinedProperties({
      ...value,
      inputFields: pipeline.inputFields ? [...pipeline.inputFields] : undefined,
      rowKeyFields: pipeline.rowKeyFields
        ? [...pipeline.rowKeyFields]
        : undefined,
      fields: [...(pipeline.fields ?? [])],
      returnFields: pipeline.returnFields?.map((field) => ({ ...field })),
      sheetContract: cloneStorageSafeSheetContract(pipeline.sheetContract),
      sheetContractErrors: pipeline.sheetContractErrors
        ? [...pipeline.sheetContractErrors]
        : undefined,
    }),
  );
}

/** Creates the complete flat V2 graph from a bounded, storage-safe pipeline. */
export function createStoredStaticPipelineV2(
  pipeline: PlayStaticPipeline | null | undefined,
): StoredStaticPipelineV2 | null | undefined {
  if (pipeline === null) return null;
  if (pipeline === undefined) return undefined;

  const bounded = truncateStaticPipelineForStorage(pipeline, {
    maxEmbeddedPlayCallPipelineDepth: Number.POSITIVE_INFINITY,
    maxStoredSubstepDepth: Number.POSITIVE_INFINITY,
  });
  if (!bounded) {
    throw new Error(
      'Static pipeline truncation unexpectedly returned no graph',
    );
  }

  const result: StoredStaticPipelineV2 = {
    version: 2,
    rootPipelineId: 'pipeline:0',
    pipelines: [],
    nodes: [],
    branches: [],
  };
  let nextPipeline = 0;
  let nextNode = 0;
  let nextBranch = 0;

  const appendPipeline = (
    current: PlayStaticPipeline,
    parentNodeId?: string,
  ): string => {
    const pipelineId = `pipeline:${nextPipeline++}`;
    result.pipelines.push({
      id: pipelineId,
      ...(parentNodeId ? { parentNodeId } : {}),
      value: storageSafePipelineValue(current),
    });

    const appendNodes = (
      entries: unknown[],
      container: StoredStaticPipelineV2Node['container'],
      parentNodeId?: string,
      branchId?: string,
    ) => {
      entries.forEach((entry, index) => {
        const record = asStaticRecord(entry);
        if (!record) return;
        const nodeId = `node:${nextNode++}`;
        const nested = asStaticRecord(record.pipeline);
        const node: StoredStaticPipelineV2Node = {
          id: nodeId,
          pipelineId,
          ...(parentNodeId ? { parentNodeId } : {}),
          ...(branchId ? { branchId } : {}),
          container,
          index,
          value: storageSafeNodeValue(record),
        };
        result.nodes.push(node);

        if (nested) {
          node.nestedPipelineId = appendPipeline(
            nested as unknown as PlayStaticPipeline,
            nodeId,
          );
        }
        if (Array.isArray(record.steps) && record.type !== 'waterfall') {
          appendNodes(record.steps, 'steps', nodeId);
        }
        if (Array.isArray(record.branches)) {
          record.branches.forEach((branch, branchIndex) => {
            const branchRecord = asStaticRecord(branch);
            if (!branchRecord || typeof branchRecord.label !== 'string') return;
            const id = `branch:${nextBranch++}`;
            result.branches.push(
              omitUndefinedProperties({
                id,
                nodeId,
                index: branchIndex,
                label: branchRecord.label,
                condition:
                  typeof branchRecord.condition === 'string'
                    ? branchRecord.condition
                    : undefined,
              }),
            );
            if (Array.isArray(branchRecord.steps)) {
              appendNodes(branchRecord.steps, 'branch_steps', nodeId, id);
            }
          });
        }
      });
    };

    appendNodes(current.stages ?? [], 'stages');
    appendNodes(current.substeps ?? [], 'substeps');
    return pipelineId;
  };

  result.rootPipelineId = appendPipeline(bounded);
  return result;
}

export const STATIC_PIPELINE_V2_INTEGRITY_ERROR_CODE =
  'PLAY_STATIC_PIPELINE_V2_INVALID';
export const STATIC_PIPELINE_V2_INTEGRITY_USER_MESSAGE =
  "This Play's stored graph is inconsistent. Republish the Play to repair it before running it again.";

export class StoredStaticPipelineV2IntegrityError extends Error {
  constructor(reason: string) {
    super(
      `${STATIC_PIPELINE_V2_INTEGRITY_ERROR_CODE}: ${STATIC_PIPELINE_V2_INTEGRITY_USER_MESSAGE} Internal reason: ${reason}.`,
    );
    this.name = 'StoredStaticPipelineV2IntegrityError';
  }
}

export function isStoredStaticPipelineV2IntegrityError(
  error: unknown,
): boolean {
  return (
    error instanceof StoredStaticPipelineV2IntegrityError ||
    (error instanceof Error &&
      error.message.includes(STATIC_PIPELINE_V2_INTEGRITY_ERROR_CODE))
  );
}

function invalidStoredStaticPipelineV2(reason: string): never {
  throw new StoredStaticPipelineV2IntegrityError(reason);
}

/**
 * Reads a present V2 envelope back into the existing in-memory pipeline shape.
 * Missing V2 is a supported legacy state. Present but malformed V2 is durable
 * corruption and fails loudly rather than running the lossy V1 projection.
 */
export function hydrateStoredStaticPipelineV2(
  value: unknown,
): PlayStaticPipeline | null {
  if (value === undefined || value === null) return null;
  const record = asStaticRecord(value);
  if (
    !record ||
    record.version !== 2 ||
    typeof record.rootPipelineId !== 'string' ||
    !Array.isArray(record.pipelines) ||
    !Array.isArray(record.nodes) ||
    !Array.isArray(record.branches)
  ) {
    return invalidStoredStaticPipelineV2('invalid envelope');
  }
  const pipelines = new Map<string, PlayStaticPipeline>();
  for (const entry of record.pipelines) {
    const pipeline = asStaticRecord(entry);
    if (!pipeline || typeof pipeline.id !== 'string') {
      return invalidStoredStaticPipelineV2('invalid pipeline record');
    }
    if (pipelines.has(pipeline.id)) {
      return invalidStoredStaticPipelineV2(
        `duplicate pipeline id ${pipeline.id}`,
      );
    }
    const pipelineValue = asStaticRecord(pipeline.value);
    if (
      !pipelineValue ||
      !Array.isArray(pipelineValue.fields) ||
      pipelineValue.fields.some((field) => typeof field !== 'string')
    ) {
      return invalidStoredStaticPipelineV2(
        `invalid pipeline value for ${pipeline.id}`,
      );
    }
    pipelines.set(pipeline.id, {
      ...(pipelineValue as unknown as PlayStaticPipeline),
      stages: [],
      substeps: [],
    });
  }
  const root = pipelines.get(record.rootPipelineId);
  if (!root) {
    return invalidStoredStaticPipelineV2(
      `missing root pipeline ${record.rootPipelineId}`,
    );
  }

  const nodes = new Map<string, Record<string, unknown>>();
  const nodeRecords = record.nodes.map((entry, position) => {
    const node = asStaticRecord(entry);
    if (!node) {
      return invalidStoredStaticPipelineV2(
        `invalid node record at position ${position}`,
      );
    }
    return node;
  });
  nodeRecords.sort((a, b) => Number(a.index) - Number(b.index));
  for (const node of nodeRecords) {
    if (
      typeof node.id !== 'string' ||
      typeof node.pipelineId !== 'string' ||
      typeof node.container !== 'string' ||
      typeof node.index !== 'number' ||
      !Number.isSafeInteger(node.index) ||
      node.index < 0
    )
      return invalidStoredStaticPipelineV2('invalid node identity');
    if (nodes.has(node.id)) {
      return invalidStoredStaticPipelineV2(`duplicate node id ${node.id}`);
    }
    const nodeValue = asStaticRecord(node.value);
    if (!nodeValue || typeof nodeValue.type !== 'string') {
      return invalidStoredStaticPipelineV2(`invalid node value for ${node.id}`);
    }
    const hydratedNode = { ...nodeValue };
    if (
      hydratedNode.type === 'dataset' ||
      hydratedNode.type === 'step_suite' ||
      hydratedNode.type === 'control_flow'
    ) {
      hydratedNode.steps = [];
    }
    nodes.set(node.id, hydratedNode);
  }
  const branches = new Map<string, Record<string, unknown>>();
  for (const entry of record.branches) {
    const branch = asStaticRecord(entry);
    if (
      !branch ||
      typeof branch.id !== 'string' ||
      typeof branch.nodeId !== 'string' ||
      typeof branch.label !== 'string' ||
      typeof branch.index !== 'number' ||
      !Number.isSafeInteger(branch.index) ||
      branch.index < 0
    )
      return invalidStoredStaticPipelineV2('invalid branch record');
    if (branches.has(branch.id)) {
      return invalidStoredStaticPipelineV2(`duplicate branch id ${branch.id}`);
    }
    branches.set(
      branch.id,
      omitUndefinedProperties({
        label: branch.label,
        condition:
          typeof branch.condition === 'string' ? branch.condition : undefined,
        steps: [] as PlayStaticSubstep[],
      }),
    );
  }
  const nestedPipelineParents = new Map<string, string>();
  for (const node of nodeRecords) {
    if (typeof node.nestedPipelineId !== 'string') continue;
    const parentPipelineId = node.pipelineId as string;
    const nestedPipelineId = node.nestedPipelineId;
    if (!pipelines.has(nestedPipelineId)) {
      return invalidStoredStaticPipelineV2(
        `node ${String(node.id)} references a missing nested pipeline`,
      );
    }
    if (nestedPipelineParents.has(nestedPipelineId)) {
      return invalidStoredStaticPipelineV2(
        `nested pipeline ${nestedPipelineId} has multiple parents`,
      );
    }
    nestedPipelineParents.set(nestedPipelineId, parentPipelineId);
    let ancestor: string | undefined = parentPipelineId;
    while (ancestor !== undefined) {
      if (ancestor === nestedPipelineId) {
        return invalidStoredStaticPipelineV2(
          `nested pipeline ${nestedPipelineId} forms a cycle`,
        );
      }
      ancestor = nestedPipelineParents.get(ancestor);
    }
  }
  for (const node of nodeRecords) {
    const nodeValue = nodes.get(node.id as string);
    const pipeline = pipelines.get(node.pipelineId as string);
    if (!nodeValue || !pipeline) {
      return invalidStoredStaticPipelineV2(
        `node ${String(node.id)} references a missing pipeline`,
      );
    }
    const container = node.container as StoredStaticPipelineV2Node['container'];
    if (container === 'stages' || container === 'substeps') {
      (pipeline[container] as PlayStaticSubstep[]).push(
        nodeValue as PlayStaticSubstep,
      );
    } else if (container === 'steps') {
      const parent = nodes.get(node.parentNodeId as string);
      if (!parent) {
        return invalidStoredStaticPipelineV2(
          `node ${String(node.id)} references a missing parent`,
        );
      }
      const steps = (parent.steps ??= []) as PlayStaticSubstep[];
      steps.push(nodeValue as PlayStaticSubstep);
    } else if (container === 'branch_steps') {
      const branch = branches.get(node.branchId as string);
      if (!branch) {
        return invalidStoredStaticPipelineV2(
          `node ${String(node.id)} references a missing branch`,
        );
      }
      (branch.steps as PlayStaticSubstep[]).push(
        nodeValue as PlayStaticSubstep,
      );
    } else {
      return invalidStoredStaticPipelineV2(
        `node ${String(node.id)} has an invalid container`,
      );
    }
    if (typeof node.nestedPipelineId === 'string') {
      const nested = pipelines.get(node.nestedPipelineId)!;
      nodeValue.pipeline = nested;
    }
  }
  for (const branchRecord of record.branches) {
    const branch = asStaticRecord(branchRecord);
    if (!branch) continue;
    const parent = nodes.get(branch.nodeId as string);
    const hydrated = branches.get(branch.id as string);
    if (!parent || !hydrated) {
      return invalidStoredStaticPipelineV2(
        `branch ${String(branch.id)} references a missing node`,
      );
    }
    const parentBranches = (parent.branches ??= []) as Record<
      string,
      unknown
    >[];
    parentBranches.push(hydrated);
  }

  // V2 deliberately omits the recursive `columns[].producers[].substep`
  // structure from node values to stay below Convex's document-depth limit.
  // Rebuild that derived provenance from each dataset's hydrated steps before
  // returning the ordinary in-memory pipeline shape.
  for (const pipeline of [...pipelines.values()].reverse()) {
    const pipelineSubsteps = pipeline.substeps ?? [];
    const compile = (substep: PlayStaticSubstep) =>
      compileStaticGraphSubstep(substep, pipelineSubsteps, []);
    pipeline.stages = (pipeline.stages ?? []).map(compile);
    pipeline.substeps = pipelineSubsteps.map(compile);
  }
  return root;
}

function storageObjectDepth(value: unknown): number {
  if (!value || typeof value !== 'object') return 0;
  const children = Array.isArray(value)
    ? value
    : Object.values(value as Record<string, unknown>);
  return (
    1 +
    children.reduce((max, child) => Math.max(max, storageObjectDepth(child)), 0)
  );
}

/**
 * V1 remains byte-for-byte shape compatible whenever it is safely shallow.
 * Only a graph that would leave too little room for its enclosing Convex
 * document receives the compact compatibility projection.
 */
export function createStaticPipelineV1CompatibilityProjection(
  pipeline: PlayStaticPipeline | null | undefined,
): PlayStaticPipeline | null | undefined {
  if (!pipeline) return pipeline;
  const stored = truncateStaticPipelineForStorage(pipeline);
  if (!stored) return stored;
  // The containing `playRevisions` / `playRuns` document adds several object
  // levels. Keep the V1 graph below this conservative bound; V2 has the full
  // graph for all current readers.
  if (storageObjectDepth(stored) <= 11) return stored;
  const shallow = (substeps: PlayStaticSubstep[] | undefined) =>
    (substeps ?? []).map((substep) => {
      const value = storageSafeNodeValue(
        substep as unknown as Record<string, unknown>,
      );
      if (
        substep.type === 'dataset' ||
        substep.type === 'step_suite' ||
        substep.type === 'control_flow'
      ) {
        value.steps = [];
      }
      return value as PlayStaticSubstep;
    });
  return {
    ...storageSafePipelineValue(stored),
    stages: shallow(stored.stages),
    substeps: shallow(stored.substeps),
  } as PlayStaticPipeline;
}

/**
 * Creates the two transport-safe fields persisted by Convex Play records.
 * Call this before crossing a Convex function boundary: the complete graph is
 * flat in V2, while V1 remains a deliberately bounded compatibility view.
 */
export function createStaticPipelineStorageFields(
  pipeline: PlayStaticPipeline | null | undefined,
): {
  staticPipeline?: PlayStaticPipeline | null;
  staticPipelineV2?: StoredStaticPipelineV2 | null;
} {
  if (pipeline === undefined) return {};
  return {
    staticPipeline: createStaticPipelineV1CompatibilityProjection(pipeline),
    staticPipelineV2: createStoredStaticPipelineV2(pipeline),
  };
}

/** Prefers complete V2. Falls back only when a legacy row has no V2 value. */
export function readStoredStaticPipeline(input: {
  staticPipeline?: unknown;
  staticPipelineV2?: unknown;
}): PlayStaticPipeline | null {
  if (input.staticPipelineV2 !== undefined && input.staticPipelineV2 !== null) {
    return hydrateStoredStaticPipelineV2(input.staticPipelineV2);
  }
  return (
    (input.staticPipeline as PlayStaticPipeline | null | undefined) ?? null
  );
}

export interface PlayStaticSourceRange {
  sourcePath?: string;
  startLine: number;
  endLine: number;
  startColumn: number;
  endColumn: number;
}

type PlayStaticSubstepMetadata = {
  conditional?: boolean;
  disabled?: boolean;
};

/**
 * One arm of a conditional `control_flow` substep — an `if`/`else if`/`else`
 * leg, a `switch` case, or a ternary branch. Carries its own steps so the graph
 * can render the conditional as a real fork instead of a flat strip.
 */
export type PlayStaticControlFlowBranch = {
  /** Short arm label, e.g. `if`, `else if`, `else`, `case 'x'`, `default`. */
  label: string;
  /** The arm's condition source text, when it has one (omitted for `else`). */
  condition?: string;
  /** Static work in this arm. Empty for log/throw/return-only arms. */
  steps: PlayStaticSubstep[];
};

export type PlayStaticSubstep = PlayStaticSubstepMetadata &
  (
    | {
        type: 'csv';
        field: string;
        path?: string;
        description?: string;
        sourceRange?: PlayStaticSourceRange;
        callDepth?: number;
        callPath?: string[];
      }
    | {
        type: 'dataset';
        field: string;
        name?: string;
        tableNamespace?: string;
        inputFields?: string[];
        rowKeyFields?: string[];
        outputFields?: string[];
        columns?: PlayStaticDatasetColumn[];
        waterfallIds?: string[];
        steps?: PlayStaticSubstep[];
        sheetContract?: PlaySheetContract | null;
        /**
         * Columns the author declared as deliberately absent from the `@mermaid`
         * diagram via `.run({ undrawnColumns: [...] })`. The docflow column
         * coverage gate reads this; nothing about execution does.
         */
        undrawnColumns?: string[];
        description?: string;
        sourceRange?: PlayStaticSourceRange;
        callDepth?: number;
        callPath?: string[];
      }
    | {
        type: 'tool';
        toolId: string;
        field: string;
        paramsSource?: string;
        sourceText?: string;
        description?: string;
        inLoop?: boolean;
        isEventWait?: boolean;
        sourceRange?: PlayStaticSourceRange;
        callDepth?: number;
        callPath?: string[];
      }
    | {
        type: 'waterfall';
        tool?: string;
        field: string;
        inLoop?: boolean;
        id?: string;
        output?: string;
        minResults?: number;
        sourceText?: string;
        steps?: Array<{
          id: string;
          kind?: 'tool' | 'code';
          toolId?: string;
          paramsSource?: string;
        }>;
        description?: string;
        sourceRange?: PlayStaticSourceRange;
        callDepth?: number;
        callPath?: string[];
      }
    | {
        type: 'step_suite';
        field: string;
        steps: PlayStaticSubstep[];
        returnSource?: string;
        description?: string;
        sourceRange?: PlayStaticSourceRange;
        callDepth?: number;
        callPath?: string[];
      }
    | {
        type: 'play_call';
        playId: string;
        execution?: 'inline' | 'child-workflow';
        timeoutMs?: number;
        hasExplicitTimeout?: boolean;
        field: string;
        inLoop?: boolean;
        pipeline?: PlayStaticPipeline | null;
        cycleDetected?: boolean;
        resolutionError?: string;
        description?: string;
        sourceRange?: PlayStaticSourceRange;
        callDepth?: number;
        callPath?: string[];
      }
    | {
        type: 'control_flow';
        kind: 'conditional' | 'loop';
        field: string;
        /** Flattened steps across every arm, in source order (back-compat). */
        steps: PlayStaticSubstep[];
        /** Discriminant source text (the `if`/ternary test, `switch` subject). */
        condition?: string;
        /** Per-arm breakdown for conditionals; omitted for loops. */
        branches?: PlayStaticControlFlowBranch[];
        description?: string;
        sourceRange?: PlayStaticSourceRange;
        callDepth?: number;
        callPath?: string[];
      }
    | {
        type: 'run_javascript';
        alias: string;
        sourceText?: string;
        description?: string;
        sourceRange?: PlayStaticSourceRange;
        callDepth?: number;
        callPath?: string[];
      }
    | {
        type: 'code';
        field: string;
        sourceText?: string;
        description?: string;
        sourceRange?: PlayStaticSourceRange;
        callDepth?: number;
        callPath?: string[];
      }
  );

export function getCompiledPipelineSubsteps(
  pipeline: PlayStaticPipeline | null | undefined,
): PlayStaticSubstep[] {
  if (!pipeline) {
    return [];
  }
  return [...(pipeline.stages ?? []), ...(pipeline.substeps ?? [])];
}

export function getTopLevelPipelineSubsteps(
  pipeline: PlayStaticPipeline | null | undefined,
): PlayStaticSubstep[] {
  return compileStaticGraph(pipeline).topLevel;
}

function getRawTopLevelPipelineSubsteps(
  pipeline: PlayStaticPipeline | null | undefined,
): PlayStaticSubstep[] {
  if (!pipeline) {
    return [];
  }
  const topLevel = [...(pipeline.stages ?? [])];
  const tableNamespace = pipeline.tableNamespace?.trim();
  if (pipeline.csvArg && !substepsContainType(topLevel, 'csv')) {
    topLevel.unshift({
      type: 'csv',
      field: 'csv',
      path: pipeline.csvArg,
      description: pipeline.csvDescription,
    });
  }
  if (tableNamespace && !substepsContainType(topLevel, 'dataset')) {
    topLevel.push({
      type: 'dataset',
      field: tableNamespace,
      tableNamespace,
      inputFields: pipeline.inputFields,
      description: pipeline.datasetDescription,
    });
  }
  if (topLevel.length > 0) {
    return topLevel;
  }
  return [...pipeline.substeps];
}

function substepsContainType(
  substeps: readonly PlayStaticSubstep[],
  type: PlayStaticSubstep['type'],
): boolean {
  for (const substep of substeps) {
    if (substep.type === type) {
      return true;
    }
    if (
      (substep.type === 'dataset' ||
        substep.type === 'step_suite' ||
        substep.type === 'control_flow') &&
      substep.steps?.length &&
      substepsContainType(substep.steps, type)
    ) {
      return true;
    }
    if (
      substep.type === 'play_call' &&
      substep.pipeline &&
      substepsContainType(getCompiledPipelineSubsteps(substep.pipeline), type)
    ) {
      return true;
    }
  }
  return false;
}

export function flattenStaticSubsteps(
  substeps: PlayStaticSubstep[],
): PlayStaticSubstep[] {
  const flattened: PlayStaticSubstep[] = [];

  for (const substep of substeps) {
    flattened.push(substep);
    if (substep.type === 'dataset' && substep.steps?.length) {
      flattened.push(...flattenStaticSubsteps(substep.steps));
      continue;
    }
    if (substep.type === 'step_suite') {
      flattened.push(...flattenStaticSubsteps(substep.steps));
      continue;
    }
    if (substep.type === 'control_flow') {
      flattened.push(...flattenStaticSubsteps(substep.steps));
      continue;
    }
    if (substep.type === 'play_call' && substep.pipeline) {
      const nestedSubsteps = getCompiledPipelineSubsteps(substep.pipeline);
      if (nestedSubsteps.length > 0) {
        flattened.push(...flattenStaticSubsteps(nestedSubsteps));
      }
    }
  }

  return flattened;
}

export function flattenStaticPipeline(
  pipeline: PlayStaticPipeline,
): PlayStaticSubstep[] {
  return flattenStaticSubsteps(getCompiledPipelineSubsteps(pipeline));
}

export function compileStaticGraph(
  pipeline: PlayStaticPipeline | null | undefined,
): PlayCompiledStaticGraph {
  const rawTopLevel = getRawTopLevelPipelineSubsteps(pipeline);
  const datasets: PlayCompiledStaticGraph['datasets'] = [];
  const topLevel = rawTopLevel.map((substep) =>
    compileStaticGraphSubstep(substep, pipeline?.substeps ?? [], datasets),
  );
  return { topLevel, datasets };
}

function compileStaticGraphSubstep(
  substep: PlayStaticSubstep,
  pipelineSubsteps: PlayStaticSubstep[],
  datasets: PlayCompiledStaticGraph['datasets'],
): PlayStaticSubstep {
  if (substep.type === 'dataset') {
    const columns = compileDatasetColumns(substep, pipelineSubsteps);
    const tableNamespace = (substep.tableNamespace ?? substep.field).trim();
    if (tableNamespace) {
      datasets.push({ tableNamespace, columns });
    }
    return {
      ...substep,
      columns,
      steps: substep.steps?.map((nested) =>
        compileStaticGraphSubstep(nested, pipelineSubsteps, datasets),
      ),
    } satisfies PlayStaticSubstep;
  }

  if (substep.type === 'step_suite' || substep.type === 'control_flow') {
    return {
      ...substep,
      steps: substep.steps.map((nested) =>
        compileStaticGraphSubstep(nested, pipelineSubsteps, datasets),
      ),
    } satisfies PlayStaticSubstep;
  }

  return substep;
}

function compileDatasetColumns(
  dataset: Extract<PlayStaticSubstep, { type: 'dataset' }>,
  pipelineSubsteps: PlayStaticSubstep[] = [],
): PlayStaticDatasetColumn[] {
  const columnsById = new Map<string, PlayStaticDatasetColumn>();
  const ensureColumn = (
    id: string,
    source: PlaySheetColumnSource,
    sqlName?: string,
  ) => {
    const trimmed = id.trim();
    if (!trimmed) return null;
    const existing = columnsById.get(trimmed);
    if (existing) {
      if (!existing.sqlName && sqlName) existing.sqlName = sqlName;
      return existing;
    }
    const column: PlayStaticDatasetColumn = {
      id: trimmed,
      source,
      ...(sqlName ? { sqlName } : {}),
      producers: [],
    };
    columnsById.set(trimmed, column);
    return column;
  };

  for (const column of dataset.sheetContract?.columns ?? []) {
    ensureColumn(column.id, column.source, column.sqlName);
  }
  for (const field of dataset.inputFields ?? []) {
    ensureColumn(field, 'input', sqlSafePlayColumnName(field));
  }
  for (const field of dataset.outputFields ?? []) {
    ensureColumn(field, 'datasetColumn', sqlSafePlayColumnName(field));
  }

  const datasetProducerSteps =
    dataset.steps && dataset.steps.length > 0
      ? dataset.steps
      : pipelineSubsteps.filter((substep) => {
          const field = fieldForColumnProducer(substep);
          return field ? (dataset.outputFields ?? []).includes(field) : false;
        });

  for (const substep of datasetProducerSteps) {
    const field = fieldForColumnProducer(substep);
    if (!field) continue;
    const column = ensureColumn(
      field,
      'datasetColumn',
      sqlSafePlayColumnName(field),
    );
    if (!column) continue;
    const producer = columnProducerFromSubstep(substep, field);
    column.producers.push(producer);
  }

  return [...columnsById.values()];
}

function fieldForColumnProducer(substep: PlayStaticSubstep): string | null {
  if ('field' in substep && typeof substep.field === 'string') {
    return substep.field.trim() || null;
  }
  if (substep.type === 'run_javascript') {
    return substep.alias.trim() || null;
  }
  return null;
}

function columnProducerFromSubstep(
  substep: PlayStaticSubstep,
  field: string,
): PlayStaticColumnProducer {
  const nestedProducerSubsteps =
    substep.type === 'control_flow' && substep.steps.length === 0
      ? (substep.branches ?? []).flatMap((branch) => branch.steps)
      : substep.type === 'step_suite' || substep.type === 'control_flow'
        ? substep.steps
        : null;
  const steps = nestedProducerSubsteps
    ? nestedProducerSubsteps
        .map((step) => {
          const stepField = fieldForColumnProducer(step) ?? field;
          return columnProducerFromSubstep(step, stepField);
        })
        .filter((producer) => producer.field.trim())
    : undefined;
  const kind: PlayStaticColumnProducerKind =
    substep.type === 'tool'
      ? 'tool'
      : substep.type === 'waterfall'
        ? 'waterfall'
        : substep.type === 'step_suite'
          ? 'stepProgram'
          : substep.type === 'control_flow'
            ? 'controlFlow'
            : substep.type === 'play_call'
              ? 'playCall'
              : 'transform';
  return {
    id: producerId(substep, field),
    kind,
    field,
    ...(substep.type === 'tool' ? { toolId: substep.toolId } : {}),
    ...(substep.type === 'play_call' ? { playId: substep.playId } : {}),
    ...(substep.conditional ? { conditional: true } : {}),
    ...(substep.disabled ? { disabled: true } : {}),
    ...(substep.sourceRange ? { sourceRange: substep.sourceRange } : {}),
    ...(steps && steps.length > 0 ? { steps } : {}),
    substep,
  };
}

function producerId(substep: PlayStaticSubstep, field: string): string {
  if (substep.type === 'tool') return `${field}:tool:${substep.toolId}`;
  if (substep.type === 'waterfall') {
    return `${field}:waterfall:${substep.id ?? substep.tool ?? 'unknown'}`;
  }
  if (substep.type === 'play_call') return `${field}:play:${substep.playId}`;
  if (substep.type === 'step_suite') return `${field}:steps`;
  if (substep.type === 'control_flow')
    return `${field}:control:${substep.kind}:${sourceRangeKey(substep.sourceRange)}`;
  if (substep.type === 'run_javascript')
    return `${field}:transform:${substep.alias}`;
  return `${field}:transform`;
}

function sourceRangeKey(
  sourceRange: PlayStaticSourceRange | undefined,
): string {
  return sourceRange
    ? `${sourceRange.startLine}:${sourceRange.startColumn}:${sourceRange.endLine}:${sourceRange.endColumn}`
    : 'unknown';
}

export function resolveSheetContractForTableNamespace(
  pipeline: PlayStaticPipeline | null | undefined,
  tableNamespace: string | null | undefined,
): PlaySheetContract | null {
  const requestedNamespace = tableNamespace?.trim();
  if (!pipeline || !requestedNamespace) {
    return null;
  }
  const normalizedNamespace = normalizeTableNamespace(requestedNamespace);
  const seen = new Set<PlayStaticPipeline>();

  const resolveFromPipeline = (
    currentPipeline: PlayStaticPipeline | null | undefined,
  ): PlaySheetContract | null => {
    if (!currentPipeline || seen.has(currentPipeline)) {
      return null;
    }
    seen.add(currentPipeline);

    const rootNamespace = currentPipeline.tableNamespace?.trim();
    if (
      rootNamespace &&
      normalizeTableNamespace(rootNamespace) === normalizedNamespace &&
      currentPipeline.sheetContract
    ) {
      return currentPipeline.sheetContract;
    }

    for (const substep of getCompiledPipelineSubsteps(currentPipeline)) {
      if (substep.type === 'dataset') {
        const substepNamespace = substep.tableNamespace?.trim();
        if (
          substepNamespace &&
          normalizeTableNamespace(substepNamespace) === normalizedNamespace &&
          substep.sheetContract
        ) {
          return substep.sheetContract;
        }
        continue;
      }

      // A ctx.dataset() inside an if/switch/loop is represented inside the
      // control-flow step's flattened `steps`. Its contract is just as durable
      // as a top-level dataset contract: runtime sheet registration resolves by
      // namespace after the branch is chosen. Do not make that registration
      // depend on which branch happened to be first in the source.
      if (
        (substep.type === 'control_flow' || substep.type === 'step_suite') &&
        substep.steps.length > 0
      ) {
        const nestedPipeline: PlayStaticPipeline = {
          stages: substep.steps,
          substeps: [],
          fields: [],
        };
        const nestedContract = resolveFromPipeline(nestedPipeline);
        if (nestedContract) {
          return nestedContract;
        }
        continue;
      }

      if (substep.type === 'play_call') {
        const nestedContract = resolveFromPipeline(substep.pipeline);
        if (nestedContract) {
          return nestedContract;
        }
      }
    }

    return null;
  };

  return resolveFromPipeline(pipeline);
}

export function resolveStaticDatasetColumnsForTableNamespace(
  pipeline: PlayStaticPipeline | null | undefined,
  tableNamespace: string | null | undefined,
): PlayStaticDatasetColumn[] {
  const requestedNamespace = tableNamespace?.trim();
  if (!pipeline || !requestedNamespace) {
    return [];
  }
  const normalizedNamespace = normalizeTableNamespace(requestedNamespace);
  const seen = new Set<PlayStaticPipeline>();

  const resolveFromPipeline = (
    currentPipeline: PlayStaticPipeline | null | undefined,
  ): PlayStaticDatasetColumn[] | null => {
    if (!currentPipeline || seen.has(currentPipeline)) {
      return null;
    }
    seen.add(currentPipeline);

    const compiled = compileStaticGraph(currentPipeline);
    const matchingDataset = compiled.datasets.find(
      (dataset) =>
        normalizeTableNamespace(dataset.tableNamespace) === normalizedNamespace,
    );
    if (matchingDataset) {
      return matchingDataset.columns;
    }

    for (const substep of getCompiledPipelineSubsteps(currentPipeline)) {
      if (substep.type === 'play_call') {
        const nestedColumns = resolveFromPipeline(substep.pipeline);
        if (nestedColumns) {
          return nestedColumns;
        }
      }
    }

    return null;
  };

  return resolveFromPipeline(pipeline) ?? [];
}

export function compileSheetContract(pipeline: PlayStaticPipeline): {
  contract: PlaySheetContract | null;
  errors: string[];
} {
  const errors = Array.isArray(pipeline.sheetContractErrors)
    ? [...pipeline.sheetContractErrors]
    : [];
  const tableNamespace =
    typeof pipeline.tableNamespace === 'string'
      ? pipeline.tableNamespace.trim()
      : '';
  if (!tableNamespace) {
    return {
      contract: null,
      errors,
    };
  }

  const columns: PlaySheetColumnContract[] = [];
  const inputFields = Array.isArray(pipeline.inputFields)
    ? pipeline.inputFields.filter((field): field is string => {
        if (typeof field === 'string') return true;
        errors.push('Sheet contract ignored a non-string input field.');
        return false;
      })
    : [];
  const rowKeyFields = Array.isArray(pipeline.rowKeyFields)
    ? pipeline.rowKeyFields.filter((field): field is string => {
        if (typeof field === 'string') return true;
        errors.push('Sheet contract ignored a non-string row key field.');
        return false;
      })
    : [];
  const fields = Array.isArray(pipeline.fields)
    ? pipeline.fields.filter((field): field is string => {
        if (typeof field === 'string') return true;
        errors.push('Sheet contract ignored a non-string output field.');
        return false;
      })
    : [];
  const substeps = Array.isArray(pipeline.substeps) ? pipeline.substeps : [];
  if (!Array.isArray(pipeline.fields)) {
    errors.push('Sheet contract could not read static output fields.');
  }
  if (!Array.isArray(pipeline.substeps)) {
    errors.push('Sheet contract could not read static substeps.');
  }

  const addColumn = (column: Omit<PlaySheetColumnContract, 'sqlName'>) => {
    if (!column.id.trim()) {
      errors.push('Sheet contract produced an empty column id.');
      return;
    }
    const existing = columns.find((candidate) => candidate.id === column.id);
    if (existing) {
      if (existing.source === 'input' && column.source === 'datasetColumn') {
        existing.source = 'datasetColumn';
        existing.field = column.field;
      }
      return;
    }
    columns.push({
      ...column,
      sqlName: sqlSafePlayColumnName(column.id),
    });
  };

  const rowKeyFieldSet = new Set(rowKeyFields);
  for (const inputField of inputFields) {
    addColumn({
      id: inputField,
      source: 'input',
      field: inputField,
      ...(rowKeyFieldSet.has(inputField) ? { isRowKey: true } : {}),
    });
  }

  for (const field of fields) {
    addColumn({ id: field, source: 'datasetColumn', field });
  }

  const processSubstep = (substep: PlayStaticSubstep) => {
    if (substep.type === 'waterfall') {
      if (!substep.id) {
        if (substep.tool) {
          return;
        }
        errors.push(
          `Sheet contract cannot compile waterfall field "${substep.field}" without a literal waterfall id.`,
        );
        return;
      }
      if (!substep.steps?.length) {
        errors.push(
          `Sheet contract cannot compile waterfall "${substep.id}" because its steps are not statically known. ` +
            'Use an inline array, a local const array, or a local no-arg function that returns an array of step("id", "tool", ...) calls.',
        );
        return;
      }
      for (const step of substep.steps) {
        addColumn({
          id: `${substep.id}.${step.id}`,
          source: 'waterfallStep',
          field: substep.field,
          waterfallId: substep.id,
          outputField: substep.output,
          outputSqlName: substep.output
            ? sqlSafePlayColumnName(substep.output)
            : undefined,
          stepId: step.id,
          toolId: step.toolId,
        });
      }
      return;
    }

    if (substep.type === 'step_suite') {
      const addStepSuiteColumns = (
        suite: Extract<PlayStaticSubstep, { type: 'step_suite' }>,
        rootField: string,
      ) => {
        for (const step of suite.steps) {
          if (!('field' in step)) {
            continue;
          }
          const stepId = step.field.startsWith(`${rootField}.`)
            ? step.field.slice(rootField.length + 1)
            : step.field;
          if (!stepId.trim()) {
            continue;
          }
          addColumn({
            id: step.field,
            source: 'waterfallStep',
            field: rootField,
            waterfallId: rootField,
            outputField: rootField,
            outputSqlName: sqlSafePlayColumnName(rootField),
            stepId,
            toolId: step.type === 'tool' ? step.toolId : undefined,
          });
          if (step.type === 'step_suite') {
            addStepSuiteColumns(step, rootField);
          }
          if (step.type === 'control_flow') {
            for (const nested of step.steps) {
              if (nested.type === 'step_suite') {
                addStepSuiteColumns(nested, rootField);
              } else {
                processSubstep(nested);
              }
            }
          }
        }
      };
      addStepSuiteColumns(substep, substep.field);
      return;
    }
    if (substep.type === 'control_flow') {
      for (const nested of substep.steps) {
        processSubstep(nested);
      }
      return;
    }
    if (substep.type === 'play_call') {
      if (substep.cycleDetected || substep.resolutionError) {
        errors.push(
          substep.resolutionError ??
            `Sheet contract cannot compile recursive child play "${substep.playId}".`,
        );
      } else if (!substep.pipeline) {
        errors.push(
          `Sheet contract cannot compile child play field "${substep.field}" until "${substep.playId}" is statically resolved.`,
        );
      } else {
        const child = compileSheetContract(substep.pipeline);
        for (const childError of child.errors) {
          errors.push(`${substep.playId}: ${childError}`);
        }
        if (child.contract) {
          for (const childColumn of child.contract.columns) {
            if (childColumn.source === 'input') continue;
            addColumn({
              id: `${substep.field}.${childColumn.id}`,
              source: 'childPlayColumn',
              parentField: substep.field,
              playId: substep.playId,
              field: childColumn.field,
              waterfallId: childColumn.waterfallId,
              outputField: childColumn.outputField,
              outputSqlName: childColumn.outputSqlName,
              stepId: childColumn.stepId,
              toolId: childColumn.toolId,
            });
          }
        }
      }
    }
  };

  for (const substep of substeps) {
    processSubstep(substep);
  }

  return {
    contract: {
      tableNamespace,
      columns,
    },
    errors: [...new Set(errors)],
  };
}