///|
// A block summary contains only locals changed within that block. Negative
// sources name a block entry; nonnegative sources retain ordinary write IDs.
// Resolve entry sources on demand instead of retaining locals × blocks states.
fn local_graph_sparse_entry_sources(
  cfg : HotCfg,
  summaries : Array[@hashmap.HashMap[Int32, Array[Int32]]],
  block_id : BlockId,
  local_id : LocalId,
  include_exceptional : Bool,
  cache : Array[@hashmap.HashMap[Int32, Array[Int32]]],
  seen : Array[Bool],
  visited : Array[BlockId],
  work : Array[BlockId],
) -> Array[Int32] {
  match cache[block_id].get(local_id) {
    Some(sources) => return sources
    None => ()
  }
  let sources : Array[Int32] = []
  work.push(block_id)
  seen[block_id] = true
  visited.push(block_id)
  while !work.is_empty() {
    let current = work.pop().unwrap()
    if current == cfg_entry_block(cfg) {
      ignore(
        local_graph_union_into(sources, [local_graph_entry_source_id(local_id)]),
      )
    }
    for pred in cfg_predecessor_edges(cfg, current) {
      if !include_exceptional && pred.kind == CfgEdgeKind::ExceptionalEdge {
        continue
      }
      match summaries[pred.block].get(local_id) {
        Some(summary) =>
          for source in summary {
            if source >= 0 {
              if !local_graph_array_contains(sources, source) {
                sources.push(source)
              }
            } else {
              let entry = -source - 1
              if !seen[entry] {
                seen[entry] = true
                visited.push(entry)
                work.push(entry)
              }
            }
          }
        None =>
          if !seen[pred.block] {
            seen[pred.block] = true
            visited.push(pred.block)
            work.push(pred.block)
          }
      }
    }
  }
  // Cache only after the complete backwards closure. A cycle encountered
  // during traversal is not evidence that its remaining entry sources vanish.
  cache[block_id].set(local_id, sources)
  for block in visited {
    seen[block] = false
  }
  visited.clear()
  sources
}

///|
fn local_graph_build_sparse_flow(
  func : HotFunc,
  cfg : HotCfg,
  include_exceptional : Bool,
  query_builds~ : Array[Int32]? = None,
) -> HotLocalGraph {
  local_graph_finish_full_flow(
    func,
    local_graph_build_sparse_rows(
      func,
      cfg,
      include_exceptional,
      query_builds~,
      full_metadata=true,
    ),
  )
}

///|
fn local_graph_build_sparse_rows(
  func : HotFunc,
  cfg : HotCfg,
  include_exceptional : Bool,
  query_builds~ : Array[Int32]? = None,
  full_metadata~ : Bool,
) -> LocalGraphFlowRows {
  let local_count = hot_local_count(func)
  let node_count = hot_node_count(func)
  let get_sources : Array[Array[Int32]] = Array::make(node_count, [])
  let recorded_sources = LocalGraphSourceRecorder::new(get_sources)
  let summaries : Array[@hashmap.HashMap[Int32, Array[Int32]]] = []
  let cache : Array[@hashmap.HashMap[Int32, Array[Int32]]] = []
  for block = 0; block < cfg_block_count(cfg); block = block + 1 {
    let entry_source = -block - 1
    let state = Array::make(local_count, [entry_source])
    // Reuse the same path-sensitive transfer for nested conditionals,
    // unknown nested regions and shared tuple evaluations as the full solver.
    let exit_state = local_graph_transfer_block(
      func,
      cfg,
      block,
      state,
      Some(recorded_sources),
      None,
    )
    let summary : @hashmap.HashMap[Int32, Array[Int32]] = @hashmap.new()
    for local_id = 0; local_id < local_count; local_id = local_id + 1 {
      let sources = exit_state[local_id]
      if sources.length() != 1 || sources[0] != entry_source {
        summary.set(local_id, sources)
      }
    }
    summaries.push(summary)
    cache.push(@hashmap.new())
  }
  let seen = Array::make(cfg_block_count(cfg), false)
  let visited : Array[BlockId] = []
  let work : Array[BlockId] = []
  match query_builds {
    Some(count) => count[0] += 1
    None => ()
  }
  let write_influences = if full_metadata {
    local_graph_empty_influences(func)
  } else {
    []
  }
  let influence_recorder = if full_metadata {
    Some(write_influences)
  } else {
    None
  }
  for node = 0; node < node_count; node = node + 1 {
    if !hot_node_is_live(func, node) ||
      !(hot_node_get(func, node).op is HotOp::LocalGet) {
      continue
    }
    let local_id = hot_node_get(func, node).imm0
    let symbolic = get_sources[node]
    get_sources[node] = []
    for source in symbolic {
      let resolved = if source >= 0 {
        [source]
      } else {
        local_graph_sparse_entry_sources(
          cfg,
          summaries,
          -source - 1,
          local_id,
          include_exceptional,
          cache,
          seen,
          visited,
          work,
        )
      }
      local_graph_record_get_sources(
        local_count,
        node,
        resolved,
        Some(recorded_sources),
        influence_recorder,
      )
    }
  }
  let write_facts = local_graph_build_write_facts(
    func,
    include_tees=full_metadata,
  )
  LocalGraphFlowRows::new(get_sources, write_influences, write_facts)
}