refactor(scope-resolution): generic orchestrator + language-agnostic phase

G-Units 6-7 of the emit-pipeline generalization plan, plus the
pipeline-phase generalization (the user's observation that the phase
itself is generic once the orchestrator is).

Changes:

- emit-core/orchestrator.ts — runScopeResolution(input, provider).
  The 180 lines of pipeline glue moved here, parametrized by
  EmitProvider. Provider supplies LanguageProvider, importEdgeReason,
  and the 6 emit-side hooks.
- emit-core/emit-provider.ts — EmitProvider gains languageProvider
  and importEdgeReason fields so the orchestrator needs nothing else.
  resolveImportTarget now takes (targetRaw, fromFile, allFilePaths).
- languages/python/emit/index.ts — pythonEmitProvider + thin
  runPythonScopeResolution wrapper. The first reference impl every
  next-language migration copies.
- emit-providers-registry.ts (NEW) — registry of per-language
  EmitProviders keyed by SupportedLanguages. Adding a language is
  one line here + the provider file.
- pipeline-phases/scope-resolution.ts (NEW) — language-agnostic phase
  iterating EMIT_PROVIDERS ∩ MIGRATED_LANGUAGES. Replaces
  pipeline-phases/python-scope.ts (deleted).
- python-scope-emit.ts deleted.
- pipeline.ts swaps pythonScopePhase → scopeResolutionPhase.

The next language migration is now: implement EmitProvider, register
it, add to MIGRATED_LANGUAGES. No new pipeline phase, no orchestrator
copy-paste. The Python migration's 700+ lines of glue collapse to
~80 lines per future language.

Verification:
- REGISTRY_PRIMARY_PYTHON=0 (legacy): 191/191.
- REGISTRY_PRIMARY_PYTHON=1 (registry): 191/191.
- Default (post MIGRATED_LANGUAGES flip): 191/191.
- tsc --noEmit clean.

gitnexus/src/core/ingestion/emit-core/emit-provider.ts

@@ -42,6 +42,7 @@ import type {
 } from 'gitnexus-shared';
 import type { KnowledgeGraph } from '../../graph/types.js';
 import type { GraphNodeLookup } from './graph-node-lookup.js';
+import { LanguageProvider } from '../language-provider.js';
 
 /** A LinearizeStrategy receives the full ancestor map so C3-style
  *  algorithms (which need to merge each parent's MRO) can implement
@@ -60,6 +61,16 @@ export interface EmitProvider {
   /** Identity for telemetry + per-language flag check. */
   readonly language: SupportedLanguages;
 
+  /** Parsing-side hook bag consumed by `extractParsedFile`. The
+   *  same `LanguageProvider` reference flows through both interfaces
+   *  to keep parsing and emit semantics in sync. */
+  readonly languageProvider: LanguageProvider;
+
+  /** Reason text on emitted IMPORTS edges. Mirrors the legacy DAG's
+   *  per-language convention so consumers asserting on reason keep
+   *  working. */
+  readonly importEdgeReason: string;
+
   // ─── Pipeline hooks ────────────────────────────────────────────────────────
 
   /**
@@ -69,8 +80,17 @@ export interface EmitProvider {
    *
    * Called once per `ParsedImport` during `finalizeScopeModel`. The
    * Python implementation wraps `resolvePythonImportTarget`.
+   *
+   * `allFilePaths` is the workspace's file set — needed by per-language
+   * resolvers that must distinguish "this module exists in the repo"
+   * from "this module is external" (Python's fallback resolver, for
+   * example).
    */
-  resolveImportTarget(targetRaw: string, fromFile: string): string | null;
+  resolveImportTarget(
+    targetRaw: string,
+    fromFile: string,
+    allFilePaths: ReadonlySet<string>,
+  ): string | null;
 
   /**
    * Per-scope binding-merge precedence. The shared finalize pass

gitnexus/src/core/ingestion/emit-core/index.ts

@@ -50,3 +50,8 @@ export { collectNamespaceTargets } from './namespace-targets.js';
 export { buildPopulatedMethodDispatch } from './method-dispatch-bridge.js';
 export type { ArityVerdict, EmitProvider, LinearizeStrategy } from './emit-provider.js';
 export { buildMro, defaultLinearize } from './build-mro.js';
+export {
+  runScopeResolution,
+  type RunScopeResolutionInput,
+  type RunScopeResolutionStats,
+} from './orchestrator.js';

gitnexus/src/core/ingestion/emit-core/orchestrator.ts

@@ -0,0 +1,161 @@
+/**
+ * `runScopeResolution` — generic registry-primary resolution
+ * orchestrator.
+ *
+ *     ParsedFile[]  (one per file via `extractParsedFile`)
+ *        │  finalizeScopeModel(  + provider hooks adapted to FinalizeHooks)
+ *        ▼
+ *     ScopeResolutionIndexes
+ *        │  resolveReferenceSites
+ *        ▼
+ *     ReferenceIndex
+ *        │  emitReceiverBoundCalls (FIRST — see Contract Invariant I1)
+ *        │  emitFreeCallFallback   (THEN)
+ *        │  emitReferencesViaLookup (LAST — uses handledSites)
+ *        │  emitImportEdges
+ *        ▼
+ *     KnowledgeGraph
+ *
+ * Per-language entry points (e.g. `runPythonScopeResolution` in
+ * `languages/python/emit/index.ts`) construct an `EmitProvider` and
+ * delegate here.
+ *
+ * Plan: `docs/plans/2026-04-20-001-refactor-emit-pipeline-generalization-plan.md`.
+ */
+
+import type { ParsedFile, RegistryProviders } from 'gitnexus-shared';
+import type { KnowledgeGraph } from '../../graph/types.js';
+import { extractParsedFile } from '../scope-extractor-bridge.js';
+import { finalizeScopeModel } from '../finalize-orchestrator.js';
+import { resolveReferenceSites, type ResolveStats } from '../resolve-references.js';
+import { buildGraphNodeLookup } from './graph-node-lookup.js';
+import { buildPopulatedMethodDispatch } from './method-dispatch-bridge.js';
+import { propagateImportedReturnTypes } from './propagate-return-types.js';
+import { emitReceiverBoundCalls } from './emit-receiver-bound.js';
+import { emitFreeCallFallback } from './emit-free-call.js';
+import { emitReferencesViaLookup } from './emit-references.js';
+import { emitImportEdges } from './emit-imports.js';
+import type { EmitProvider } from './emit-provider.js';
+
+export interface RunScopeResolutionInput {
+  readonly graph: KnowledgeGraph;
+  readonly files: readonly { readonly path: string; readonly content: string }[];
+  readonly onWarn?: (message: string) => void;
+}
+
+export interface RunScopeResolutionStats {
+  readonly filesProcessed: number;
+  readonly filesSkipped: number;
+  readonly importsEmitted: number;
+  readonly resolve: ResolveStats;
+  readonly referenceEdgesEmitted: number;
+  readonly referenceSkipped: number;
+}
+
+export function runScopeResolution(
+  input: RunScopeResolutionInput,
+  provider: EmitProvider,
+): RunScopeResolutionStats {
+  const { graph, files } = input;
+  const onWarn = input.onWarn ?? (() => {});
+
+  // ── Phase 1: extract each file → ParsedFile ────────────────────────────
+  const parsedFiles: ParsedFile[] = [];
+  let filesSkipped = 0;
+  for (const file of files) {
+    const parsed = extractParsedFile(provider.languageProvider, file.content, file.path, onWarn);
+    if (parsed === undefined) {
+      filesSkipped++;
+      continue;
+    }
+    provider.populateOwners(parsed);
+    parsedFiles.push(parsed);
+  }
+
+  if (parsedFiles.length === 0) {
+    return {
+      filesProcessed: 0,
+      filesSkipped,
+      importsEmitted: 0,
+      resolve: { sitesProcessed: 0, referencesEmitted: 0, unresolved: 0 },
+      referenceEdgesEmitted: 0,
+      referenceSkipped: 0,
+    };
+  }
+
+  // ── Phase 2: finalize → ScopeResolutionIndexes ─────────────────────────
+  const allFilePaths = new Set(parsedFiles.map((f) => f.filePath));
+  const nodeLookup = buildGraphNodeLookup(graph);
+  const mroByClassDefId = provider.buildMro(graph, parsedFiles, nodeLookup);
+
+  const indexes = finalizeScopeModel(parsedFiles, {
+    hooks: {
+      resolveImportTarget: (targetRaw, fromFile) =>
+        provider.resolveImportTarget(targetRaw, fromFile, allFilePaths),
+      mergeBindings: (existing, incoming, scopeId) =>
+        provider.mergeBindings(existing, incoming, scopeId),
+    },
+  });
+
+  // Stitch the MRO into the finalized indexes (same pattern as before
+  // generalization — finalizeScopeModel builds an empty
+  // MethodDispatchIndex by design).
+  (indexes as { methodDispatch: typeof indexes.methodDispatch }).methodDispatch =
+    buildPopulatedMethodDispatch(mroByClassDefId);
+
+  // Cross-file return-type propagation (Contract Invariant I3 timing:
+  // after finalize, before resolve).
+  if (provider.propagatesReturnTypesAcrossImports !== false) {
+    propagateImportedReturnTypes(parsedFiles, indexes);
+  }
+
+  // ── Phase 3: resolve references via Registry.lookup ────────────────────
+  const registryProviders: RegistryProviders = {
+    arityCompatibility: provider.arityCompatibility,
+  };
+  const { referenceIndex, stats: resolveStats } = resolveReferenceSites({
+    scopes: indexes,
+    providers: registryProviders,
+  });
+
+  // ── Phase 4: emit graph edges (LOAD-BEARING ORDER — see I1) ────────────
+  const handledSites = new Set<string>();
+  const receiverExtras = emitReceiverBoundCalls(
+    graph,
+    indexes,
+    parsedFiles,
+    nodeLookup,
+    handledSites,
+    provider,
+  );
+  const freeCallExtras = emitFreeCallFallback(
+    graph,
+    indexes,
+    parsedFiles,
+    nodeLookup,
+    referenceIndex,
+    handledSites,
+  );
+  const { emitted, skipped } = emitReferencesViaLookup(
+    graph,
+    indexes,
+    referenceIndex,
+    nodeLookup,
+    handledSites,
+  );
+  const importsEmitted = emitImportEdges(
+    graph,
+    indexes.imports,
+    indexes.scopeTree,
+    provider.importEdgeReason,
+  );
+
+  return {
+    filesProcessed: parsedFiles.length,
+    filesSkipped,
+    importsEmitted,
+    resolve: resolveStats,
+    referenceEdgesEmitted: emitted + receiverExtras + freeCallExtras,
+    referenceSkipped: skipped,
+  };
+}

gitnexus/src/core/ingestion/emit-providers-registry.ts

@@ -0,0 +1,27 @@
+/**
+ * Per-language `EmitProvider` registry — the lookup the generic
+ * `scopeResolutionPhase` uses to pick the right provider for each
+ * migrated language.
+ *
+ * Adding a language is two lines: implement an `EmitProvider` in
+ * `languages/<lang>/emit/index.ts` and register it here. The phase
+ * picks it up automatically — no workflow changes, no per-language
+ * pipeline phase file.
+ */
+
+import { SupportedLanguages } from 'gitnexus-shared';
+import type { EmitProvider } from './emit-core/emit-provider.js';
+import { pythonEmitProvider } from './languages/python/emit/index.js';
+
+/** Map of `SupportedLanguages` → `EmitProvider`. The phase iterates
+ *  this map intersected with `MIGRATED_LANGUAGES` (the per-language
+ *  flag set) so adding a provider here without flipping the flag is
+ *  safe — the provider sits idle until the language is migrated. */
+export const EMIT_PROVIDERS: ReadonlyMap<SupportedLanguages, EmitProvider> = new Map<
+  SupportedLanguages,
+  EmitProvider
+>([[SupportedLanguages.Python, pythonEmitProvider]]);
+
+export function getEmitProvider(lang: SupportedLanguages): EmitProvider | undefined {
+  return EMIT_PROVIDERS.get(lang);
+}

gitnexus/src/core/ingestion/languages/python/emit/index.ts

@@ -0,0 +1,89 @@
+/**
+ * Python `EmitProvider` and the `runPythonScopeResolution` entry point.
+ *
+ * The provider is a thin wiring object — Python's specific bits
+ * (super recognizer, LEGB merge precedence, Python's relative-import
+ * resolver, the simplified MRO walk) plug into the generic
+ * `runScopeResolution` orchestrator from `emit-core/`.
+ *
+ * Migration reference: when bringing up the next language
+ * (TypeScript / Java / Kotlin / Ruby), copy this file's structure —
+ * implement the 6 required `EmitProvider` fields, optionally toggle
+ * the 2 booleans, and call `runScopeResolution(input, provider)`.
+ */
+
+import type { ParsedFile, Scope, WorkspaceIndex } from 'gitnexus-shared';
+import { SupportedLanguages } from 'gitnexus-shared';
+import {
+  buildMro,
+  defaultLinearize,
+  populateClassOwnedMembers,
+  runScopeResolution,
+  type EmitProvider,
+  type RunScopeResolutionInput,
+  type RunScopeResolutionStats,
+} from '../../../emit-core/index.js';
+import { pythonProvider } from '../../python.js';
+import {
+  pythonArityCompatibility,
+  pythonMergeBindings,
+  resolvePythonImportTarget,
+  type PythonResolveContext,
+} from '../index.js';
+
+const pythonEmitProvider: EmitProvider = {
+  language: SupportedLanguages.Python,
+  languageProvider: pythonProvider,
+  importEdgeReason: 'python-scope: import',
+
+  resolveImportTarget: (targetRaw, fromFile, allFilePaths) => {
+    // PythonResolveContext expects a mutable Set; orchestrator hands us
+    // a ReadonlySet — safe to widen since the resolver only reads.
+    const ws: PythonResolveContext = {
+      fromFile,
+      allFilePaths: allFilePaths as Set<string>,
+    };
+    return resolvePythonImportTarget(
+      { kind: 'named', localName: '_', importedName: '_', targetRaw },
+      ws as unknown as WorkspaceIndex,
+    );
+  },
+
+  // Python LEGB precedence: local > import/namespace/reexport > wildcard.
+  mergeBindings: (existing, incoming, scopeId) => {
+    // pythonMergeBindings(scope, bindings) only consults BindingRef.origin
+    // for tier ordering, not scope.kind. A shape-stub satisfies the type
+    // contract without falsifying behavior. Widen the readonly result
+    // to a mutable BindingRef[] for the orchestrator's hook signature.
+    const fakeScope = { id: scopeId } as unknown as Scope;
+    return [...pythonMergeBindings(fakeScope, [...existing, ...incoming])];
+  },
+
+  // Adapter: pythonArityCompatibility predates RegistryProviders and
+  // uses (def, callsite). Contract is (callsite, def).
+  arityCompatibility: (callsite, def) => pythonArityCompatibility(def, callsite),
+
+  buildMro: (graph, parsedFiles, nodeLookup) =>
+    buildMro(graph, parsedFiles, nodeLookup, defaultLinearize),
+
+  populateOwners: (parsed: ParsedFile) => populateClassOwnedMembers(parsed),
+
+  isSuperReceiver: (text) => /^super\s*\(/.test(text),
+
+  // Python is dynamically typed — field-fallback heuristic on, return-
+  // type propagation across imports on. Both default to true; listed
+  // explicitly here for documentation.
+  fieldFallbackOnMethodLookup: true,
+  propagatesReturnTypesAcrossImports: true,
+};
+
+export { pythonEmitProvider };
+
+export interface RunPythonScopeResolutionInput extends RunScopeResolutionInput {}
+export interface RunPythonScopeResolutionStats extends RunScopeResolutionStats {}
+
+export function runPythonScopeResolution(
+  input: RunPythonScopeResolutionInput,
+): RunPythonScopeResolutionStats {
+  return runScopeResolution(input, pythonEmitProvider);
+}

gitnexus/src/core/ingestion/pipeline-phases/index.ts

@@ -16,7 +16,7 @@ export { routesPhase, type RoutesOutput, type RouteEntry } from './routes.js';
 export { toolsPhase, type ToolsOutput, type ToolDef } from './tools.js';
 export { ormPhase, type ORMOutput } from './orm.js';
 export { crossFilePhase, type CrossFileOutput } from './cross-file.js';
-export { pythonScopePhase, type PythonScopeOutput } from './python-scope.js';
+export { scopeResolutionPhase, type ScopeResolutionOutput } from './scope-resolution.js';
 export { mroPhase, type MROOutput } from './mro.js';
 export { communitiesPhase, type CommunitiesOutput } from './communities.js';
 export { processesPhase, type ProcessesOutput } from './processes.js';

gitnexus/src/core/ingestion/pipeline-phases/python-scope.ts

@@ -1,113 +0,0 @@
-/**
- * Phase: pythonScope
- *
- * Registry-primary resolution for Python files (RFC #909 Ring 4).
- *
- * Gated by `REGISTRY_PRIMARY_PYTHON=1` (via `isRegistryPrimary`). When
- * the flag is OFF (default), this phase is a no-op: the legacy paths in
- * `import-processor.ts` and `call-processor.ts` continue to handle
- * Python imports + calls and this phase contributes nothing to the
- * graph.
- *
- * When the flag is ON, this phase:
- *   1. Reads every `.py` file in the workspace.
- *   2. Drives the scope-based pipeline end-to-end (extract → finalize →
- *      resolve → emit) via `runPythonScopeResolution`.
- *   3. Emits IMPORTS / CALLS / ACCESSES / INHERITS / USES edges directly.
- *
- * Pairs with the matching gates in `import-processor.ts` and
- * `call-processor.ts` that skip Python files when this phase is active —
- * so we don't double-emit edges from both code paths.
- *
- * @deps    parse  (needs Symbol nodes already in the graph so emit-references
- *                  can attach edges to existing Function/Method/Class nodes)
- * @reads   scannedFiles
- * @writes  graph (IMPORTS, CALLS, ACCESSES, INHERITS, USES)
- */
-
-import type { PipelinePhase, PipelineContext, PhaseResult } from './types.js';
-import { getPhaseOutput } from './types.js';
-import type { StructureOutput } from './structure.js';
-import { isRegistryPrimary } from '../registry-primary-flag.js';
-import { SupportedLanguages, getLanguageFromFilename } from 'gitnexus-shared';
-import { readFileContents } from '../filesystem-walker.js';
-import { runPythonScopeResolution } from '../python-scope-emit.js';
-import { isDev } from '../utils/env.js';
-
-export interface PythonScopeOutput {
-  /** True when the flag was on and the phase actually ran. */
-  readonly ran: boolean;
-  /** Python files seen by the phase. `0` when `ran === false`. */
-  readonly filesProcessed: number;
-  /** IMPORTS edges emitted by this phase. */
-  readonly importsEmitted: number;
-  /** Reference (CALLS/ACCESSES/INHERITS/USES) edges emitted. */
-  readonly referenceEdgesEmitted: number;
-}
-
-const NOOP_OUTPUT: PythonScopeOutput = Object.freeze({
-  ran: false,
-  filesProcessed: 0,
-  importsEmitted: 0,
-  referenceEdgesEmitted: 0,
-});
-
-export const pythonScopePhase: PipelinePhase<PythonScopeOutput> = {
-  name: 'pythonScope',
-  // Depends on `parse` because emit-references attaches edges to
-  // already-existing Symbol nodes (Function/Method/Class). The legacy
-  // `parse` phase still creates those nodes; we only replace the
-  // import + call resolution layer.
-  deps: ['parse', 'structure'],
-
-  async execute(
-    ctx: PipelineContext,
-    deps: ReadonlyMap<string, PhaseResult<unknown>>,
-  ): Promise<PythonScopeOutput> {
-    if (!isRegistryPrimary(SupportedLanguages.Python)) {
-      return NOOP_OUTPUT;
-    }
-
-    const { scannedFiles } = getPhaseOutput<StructureOutput>(deps, 'structure');
-
-    // Only `.py` files; the per-language flag scopes by extension via
-    // `getLanguageFromFilename`.
-    const pythonScanned = scannedFiles.filter(
-      (f) => getLanguageFromFilename(f.path) === SupportedLanguages.Python,
-    );
-    if (pythonScanned.length === 0) return NOOP_OUTPUT;
-
-    // Read source for every Python file. `runPythonScopeResolution`
-    // re-parses each file via `pythonProvider.emitScopeCaptures`; we
-    // accept the duplicate parse cost in this first cut and can revisit
-    // by sharing `astCache` across phases if needed.
-    const filePaths = pythonScanned.map((f) => f.path);
-    const contents = await readFileContents(ctx.repoPath, filePaths);
-    const files: { path: string; content: string }[] = [];
-    for (const fp of filePaths) {
-      const content = contents.get(fp);
-      if (content !== undefined) files.push({ path: fp, content });
-    }
-
-    const stats = runPythonScopeResolution({
-      graph: ctx.graph,
-      files,
-      onWarn: (msg) => {
-        if (isDev) console.warn(`[python-scope] ${msg}`);
-      },
-    });
-
-    if (isDev) {
-      console.log(
-        `🐍 python-scope: ${stats.filesProcessed} files → ${stats.importsEmitted} IMPORTS + ${stats.referenceEdgesEmitted} reference edges (${stats.resolve.unresolved} unresolved sites, ${stats.referenceSkipped} skipped)`,
-      );
-    }
-
-    return {
-      ran: true,
-      filesProcessed: stats.filesProcessed,
-      importsEmitted: stats.importsEmitted,
-      referenceEdgesEmitted: stats.referenceEdgesEmitted,
-    };
-  },
-};

gitnexus/src/core/ingestion/pipeline-phases/scope-resolution.ts

@@ -0,0 +1,147 @@
+/**
+ * Phase: scopeResolution
+ *
+ * Generic registry-primary resolution phase (RFC #909 Ring 4).
+ *
+ * For every language in `MIGRATED_LANGUAGES` (per-language flag set)
+ * whose provider is registered in `EMIT_PROVIDERS`:
+ *   1. Filter scanned files by language extension.
+ *   2. Read file contents.
+ *   3. Drive the scope-based pipeline end-to-end via the generic
+ *      `runScopeResolution(input, provider)` orchestrator.
+ *   4. Emit IMPORTS / CALLS / ACCESSES / INHERITS / USES edges.
+ *
+ * Pairs with the per-language gates in `import-processor.ts` and
+ * `call-processor.ts` that skip files when their language is registry-
+ * primary, so we don't double-emit edges from both code paths.
+ *
+ * Adding a language is two changes:
+ *   - Implement `EmitProvider` in `languages/<lang>/emit/index.ts`
+ *     and register it in `emit-providers-registry.ts`.
+ *   - Add the language to `MIGRATED_LANGUAGES` in
+ *     `registry-primary-flag.ts`.
+ *
+ * @deps    parse  (needs Symbol nodes already in the graph so emit-references
+ *                  can attach edges to existing Function/Method/Class nodes)
+ * @reads   scannedFiles
+ * @writes  graph (IMPORTS, CALLS, ACCESSES, INHERITS, USES)
+ */
+
+import type { PipelinePhase, PipelineContext, PhaseResult } from './types.js';
+import { getPhaseOutput } from './types.js';
+import type { StructureOutput } from './structure.js';
+import { isRegistryPrimary } from '../registry-primary-flag.js';
+import { SupportedLanguages, getLanguageFromFilename } from 'gitnexus-shared';
+import { readFileContents } from '../filesystem-walker.js';
+import { runScopeResolution } from '../emit-core/index.js';
+import { EMIT_PROVIDERS } from '../emit-providers-registry.js';
+import { isDev } from '../utils/env.js';
+
+export interface ScopeResolutionOutput {
+  /** True when at least one language ran. */
+  readonly ran: boolean;
+  /** Files seen across all languages. `0` when `ran === false`. */
+  readonly filesProcessed: number;
+  /** IMPORTS edges emitted across all languages. */
+  readonly importsEmitted: number;
+  /** Reference (CALLS / ACCESSES / INHERITS / USES) edges emitted. */
+  readonly referenceEdgesEmitted: number;
+  /** Per-language breakdown for telemetry / shadow-parity. */
+  readonly perLanguage: ReadonlyMap<
+    SupportedLanguages,
+    {
+      readonly filesProcessed: number;
+      readonly importsEmitted: number;
+      readonly referenceEdgesEmitted: number;
+    }
+  >;
+}
+
+const NOOP_OUTPUT: ScopeResolutionOutput = Object.freeze({
+  ran: false,
+  filesProcessed: 0,
+  importsEmitted: 0,
+  referenceEdgesEmitted: 0,
+  perLanguage: new Map(),
+});
+
+export const scopeResolutionPhase: PipelinePhase<ScopeResolutionOutput> = {
+  name: 'scopeResolution',
+  // Depends on `parse` because emit-references attaches edges to
+  // already-existing Symbol nodes (Function/Method/Class). The legacy
+  // `parse` phase still creates those nodes; we only replace the
+  // import + call resolution layer.
+  deps: ['parse', 'structure'],
+
+  async execute(
+    ctx: PipelineContext,
+    deps: ReadonlyMap<string, PhaseResult<unknown>>,
+  ): Promise<ScopeResolutionOutput> {
+    const { scannedFiles } = getPhaseOutput<StructureOutput>(deps, 'structure');
+
+    let totalFiles = 0;
+    let totalImports = 0;
+    let totalRefs = 0;
+    let anyRan = false;
+    const perLanguage = new Map<
+      SupportedLanguages,
+      {
+        readonly filesProcessed: number;
+        readonly importsEmitted: number;
+        readonly referenceEdgesEmitted: number;
+      }
+    >();
+
+    for (const [lang, provider] of EMIT_PROVIDERS) {
+      if (!isRegistryPrimary(lang)) continue;
+
+      const langFiles = scannedFiles.filter((f) => getLanguageFromFilename(f.path) === lang);
+      if (langFiles.length === 0) continue;
+
+      const filePaths = langFiles.map((f) => f.path);
+      const contents = await readFileContents(ctx.repoPath, filePaths);
+      const files: { path: string; content: string }[] = [];
+      for (const fp of filePaths) {
+        const content = contents.get(fp);
+        if (content !== undefined) files.push({ path: fp, content });
+      }
+
+      const stats = runScopeResolution(
+        {
+          graph: ctx.graph,
+          files,
+          onWarn: (msg) => {
+            if (isDev) console.warn(`[scope-resolution:${lang}] ${msg}`);
+          },
+        },
+        provider,
+      );
+
+      anyRan = true;
+      totalFiles += stats.filesProcessed;
+      totalImports += stats.importsEmitted;
+      totalRefs += stats.referenceEdgesEmitted;
+      perLanguage.set(lang, {
+        filesProcessed: stats.filesProcessed,
+        importsEmitted: stats.importsEmitted,
+        referenceEdgesEmitted: stats.referenceEdgesEmitted,
+      });
+
+      if (isDev) {
+        console.log(
+          `🐍 scope-resolution:${lang}: ${stats.filesProcessed} files → ${stats.importsEmitted} IMPORTS + ${stats.referenceEdgesEmitted} reference edges (${stats.resolve.unresolved} unresolved sites, ${stats.referenceSkipped} skipped)`,
+        );
+      }
+    }
+
+    if (!anyRan) return NOOP_OUTPUT;
+
+    return {
+      ran: true,
+      filesProcessed: totalFiles,
+      importsEmitted: totalImports,
+      referenceEdgesEmitted: totalRefs,
+      perLanguage,
+    };
+  },
+};

gitnexus/src/core/ingestion/pipeline.ts

@@ -30,7 +30,7 @@ import {
   toolsPhase,
   ormPhase,
   crossFilePhase,
-  pythonScopePhase,
+  scopeResolutionPhase,
   mroPhase,
   communitiesPhase,
   processesPhase,
@@ -81,7 +81,7 @@ function buildPhaseList(options?: PipelineOptions): PipelinePhase[] {
     toolsPhase,
     ormPhase,
     crossFilePhase,
-    pythonScopePhase,
+    scopeResolutionPhase,
   ];
 
   if (!options?.skipGraphPhases) {

gitnexus/src/core/ingestion/python-scope-emit.ts

@@ -1,249 +0,0 @@
-/**
- * `runPythonScopeResolution` — drive the registry-primary resolution
- * pipeline end-to-end for the Python files in a workspace and emit
- * graph edges (RFC #909 Ring 3 — Python migration).
- *
- *     ParsedFile[]  (one per .py via `extractParsedFile`)
- *        │  finalizeScopeModel(  + Python hooks adapted to FinalizeHooks)
- *        ▼
- *     ScopeResolutionIndexes
- *        │  resolveReferenceSites
- *        ▼
- *     ReferenceIndex
- *        │  emitReferencesViaLookup (shared — emit-core)
- *        │  + emitReceiverBoundCalls (Python-specific; moves to
- *        │     languages/python/emit/ in Unit 11)
- *        │  + emitImportEdges (shared — emit-core)
- *        ▼
- *     KnowledgeGraph
- *
- * The orchestrator is the public seam between the gitnexus pipeline and
- * the language-agnostic scope-resolution machinery in `gitnexus-shared`.
- * It wires the Python provider's hooks into `FinalizeOrchestratorOptions`
- * and threads the workspace index through the import-target resolver.
- *
- * Gating lives in the pipeline phase (`pipeline-phases/python-scope.ts`),
- * not here — this function is "what to do" once we've decided to do it.
- */
-
-import type { ParsedFile, RegistryProviders, Scope, WorkspaceIndex } from 'gitnexus-shared';
-import type { KnowledgeGraph } from '../graph/types.js';
-import { extractParsedFile } from './scope-extractor-bridge.js';
-import { finalizeScopeModel } from './finalize-orchestrator.js';
-import { resolveReferenceSites, type ResolveStats } from './resolve-references.js';
-import { pythonProvider } from './languages/python.js';
-import {
-  pythonArityCompatibility,
-  pythonMergeBindings,
-  resolvePythonImportTarget,
-  type PythonResolveContext,
-} from './languages/python/index.js';
-import {
-  buildGraphNodeLookup,
-  buildMro,
-  buildPopulatedMethodDispatch,
-  defaultLinearize,
-  emitFreeCallFallback,
-  emitImportEdges,
-  emitReceiverBoundCalls,
-  emitReferencesViaLookup,
-  populateClassOwnedMembers,
-  propagateImportedReturnTypes,
-  type EmitProvider,
-} from './emit-core/index.js';
-import { SupportedLanguages } from 'gitnexus-shared';
-
-// ─── Public API ─────────────────────────────────────────────────────────────
-
-export interface RunPythonScopeResolutionInput {
-  readonly graph: KnowledgeGraph;
-  readonly files: readonly { readonly path: string; readonly content: string }[];
-  /** Optional warning sink (e.g. for telemetry). Failures per-file are non-fatal. */
-  readonly onWarn?: (message: string) => void;
-}
-
-export interface RunPythonScopeResolutionStats {
-  readonly filesProcessed: number;
-  readonly filesSkipped: number;
-  readonly importsEmitted: number;
-  readonly resolve: ResolveStats;
-  readonly referenceEdgesEmitted: number;
-  readonly referenceSkipped: number;
-}
-
-/**
- * Run the full registry-primary resolution path for `files` and emit the
- * resulting CALLS / ACCESSES / INHERITS / USES / IMPORTS edges into
- * `graph`. Caller is responsible for ensuring `files` are Python only.
- *
- * Returns telemetry; never throws on per-file failures (warnings flow
- * through `onWarn`).
- */
-export function runPythonScopeResolution(
-  input: RunPythonScopeResolutionInput,
-): RunPythonScopeResolutionStats {
-  const { graph, files } = input;
-  const onWarn = input.onWarn ?? (() => {});
-
-  // ── Phase 1: extract each file → ParsedFile ─────────────────────────────
-  const parsedFiles: ParsedFile[] = [];
-  let filesSkipped = 0;
-  for (const file of files) {
-    const parsed = extractParsedFile(pythonProvider, file.content, file.path, onWarn);
-    if (parsed === undefined) {
-      filesSkipped++;
-      continue;
-    }
-    populateClassOwnedMembers(parsed);
-    parsedFiles.push(parsed);
-  }
-
-  if (parsedFiles.length === 0) {
-    return {
-      filesProcessed: 0,
-      filesSkipped,
-      importsEmitted: 0,
-      resolve: { sitesProcessed: 0, referencesEmitted: 0, unresolved: 0 },
-      referenceEdgesEmitted: 0,
-      referenceSkipped: 0,
-    };
-  }
-
-  // ── Phase 2: finalize → ScopeResolutionIndexes ─────────────────────────
-  const allFilePaths = new Set(parsedFiles.map((f) => f.filePath));
-  // Pre-build a graph-node lookup (used both for MRO bridging and for
-  // edge emission below). EXTENDS edges already in the graph (from the
-  // legacy heritage processor in `parse`) drive the MRO chain — we
-  // mirror them into a `MethodDispatchIndex` so receiver-typed
-  // resolution can walk inherited methods.
-  const nodeLookup = buildGraphNodeLookup(graph);
-  const mroByClassDefId = buildMro(graph, parsedFiles, nodeLookup, defaultLinearize);
-
-  const indexes = finalizeScopeModel(parsedFiles, {
-    hooks: {
-      // Adapter: shared `finalize()` calls `resolveImportTarget(targetRaw,
-      // fromFile, ws)` with `targetRaw` already extracted; the Python
-      // provider's signature takes a synthetic `ParsedImport`. Wrap it so
-      // the hook contract is satisfied without leaking provider internals.
-      resolveImportTarget: (targetRaw, fromFile) => {
-        const ws: PythonResolveContext = { fromFile, allFilePaths };
-        return resolvePythonImportTarget(
-          { kind: 'named', localName: '_', importedName: '_', targetRaw },
-          ws as unknown as WorkspaceIndex,
-        );
-      },
-      // Python LEGB precedence: local > import/namespace/reexport > wildcard.
-      mergeBindings: (existing, incoming, scopeId) => {
-        // `pythonMergeBindings(scope, bindings)` only consults
-        // `BindingRef.origin` for tier ordering, not `scope.kind`. A
-        // shape-stub satisfies the type contract without falsifying
-        // behavior.
-        const fakeScope = { id: scopeId } as unknown as Scope;
-        return pythonMergeBindings(fakeScope, [...existing, ...incoming]);
-      },
-    },
-  });
-
-  // Stitch the MRO into the finalized indexes. `finalizeScopeModel`
-  // builds an empty MethodDispatchIndex (the comment in
-  // `finalize-orchestrator.ts:124-129` notes this is a known gap); we
-  // overwrite with a populated index that wraps the same shape.
-  (indexes as { methodDispatch: typeof indexes.methodDispatch }).methodDispatch =
-    buildPopulatedMethodDispatch(mroByClassDefId);
-
-  // Propagate return-type typeBindings across imports. The shared
-  // finalize pass copies callable bindings (`from x import f` puts
-  // `f` in the importer's bindings), but typeBindings stay file-local.
-  // Without this step, `u = get_user(); u.save()` works only when
-  // get_user is in the same file as the call. Done as a post-finalize
-  // mutation since `Scope.typeBindings` is a plain Map (per
-  // `draftToScope` line 302).
-  propagateImportedReturnTypes(parsedFiles, indexes);
-
-  // ── Phase 3: resolve references via Registry.lookup ─────────────────────
-  const providers: RegistryProviders = {
-    // The Python provider's `arityCompatibility` predates the
-    // RegistryProviders contract and uses `(def, callsite)` argument
-    // order. The contract is `(callsite, def)`. Adapt at the boundary
-    // so the provider source stays untouched.
-    arityCompatibility: (callsite, def) => pythonArityCompatibility(def, callsite),
-  };
-  const { referenceIndex, stats: resolveStats } = resolveReferenceSites({
-    scopes: indexes,
-    providers,
-  });
-
-  // ── Phase 4: emit graph edges ───────────────────────────────────────────
-  // Order matters: run the Python-specific receiver-bound and free-call
-  // passes FIRST so they record (filePath, line, col) keys for sites
-  // they emit edges for. The shared resolver then skips those sites in
-  // `emitReferencesViaLookup` so its potentially-wrong fallback (e.g.
-  // resolving `app_metrics.get_metrics()` to a same-named local function
-  // instead of the namespace target) doesn't fight the precise emission.
-  const handledSites = new Set<string>();
-  const receiverExtras = emitReceiverBoundCalls(
-    graph,
-    indexes,
-    parsedFiles,
-    nodeLookup,
-    handledSites,
-    pythonEmitProviderInline,
-  );
-  const freeCallExtras = emitFreeCallFallback(
-    graph,
-    indexes,
-    parsedFiles,
-    nodeLookup,
-    referenceIndex,
-    handledSites,
-  );
-
-  // The shared `emit-references.ts` emits edges between
-  // `SymbolDefinition.nodeId` values, which use the scope-extractor's
-  // `def:<file>#<line>:<col>:<type>:<name>` format. The CLI's existing
-  // graph nodes (created by `parsing-processor.ts`) use the legacy
-  // `<Type>:<file>:<qualifiedName>` ID format. Bridging is required so
-  // edges actually link to existing graph nodes.
-  const { emitted, skipped } = emitReferencesViaLookup(
-    graph,
-    indexes,
-    referenceIndex,
-    nodeLookup,
-    handledSites,
-  );
-
-  // IMPORTS edges: the scope-resolution path now owns Python file→file
-  // IMPORTS edge emission when `REGISTRY_PRIMARY_PYTHON=1`. The legacy
-  // `processImports` path still runs (heritage needs its `importMap`
-  // population for `ctx.resolve`), but import-processor's graph edge
-  // emission is gated per-language in `createImportEdgeHelpers` so
-  // Python no longer double-emits.
-  const importsEmitted = emitImportEdges(
-    graph,
-    indexes.imports,
-    indexes.scopeTree,
-    'python-scope: import',
-  );
-
-  return {
-    filesProcessed: parsedFiles.length,
-    filesSkipped,
-    importsEmitted,
-    resolve: resolveStats,
-    referenceEdgesEmitted: emitted + receiverExtras + freeCallExtras,
-    referenceSkipped: skipped,
-  };
-}
-
-/** Minimal `EmitProvider` carrying only the hooks the receiver-bound
- *  pass currently consults. The full provider (with mergeBindings,
- *  resolveImportTarget, arityCompatibility, buildMro, populateOwners)
- *  lands in G-Unit 6 when it moves to `languages/python/emit/`. */
-const pythonEmitProviderInline: Pick<
-  EmitProvider,
-  'language' | 'isSuperReceiver' | 'fieldFallbackOnMethodLookup'
-> = {
-  language: SupportedLanguages.Python,
-  isSuperReceiver: (text) => /^super\s*\(/.test(text),
-  fieldFallbackOnMethodLookup: true,
-};