fix: prevent phantom cycles from package-module name collisions

graphify/extractors/resolution.py

@@ -1433,6 +1433,18 @@ def exported_candidates(
         )
 
     # #1146: emit file-to-file imports_from edges for package-form submodule imports.
+    # #3777: retract provisional `imports_from` AST edges whose package ID coincided
+    # with a same-named sibling module file (e.g. `from nettacker import logger`
+    # emitting a provisional edge to `nettacker.py`).
+    provisional_by_loc: dict[tuple[Path, str], list[dict]] = {}
+    for edge in edges:
+        if edge.get("relation") == "imports_from" and edge.get("context") == "import":
+            src = _js_source_path(str(edge.get("source_file", "")), root)
+            loc = edge.get("source_location")
+            if src is not None and loc:
+                provisional_by_loc.setdefault((src, loc), []).append(edge)
+
+    retracted_edge_ids: set[int] = set()
     for from_path, to_path, line, local_name in facts.module_imports:
         try:
             from_rel = from_path.relative_to(root)
@@ -1441,11 +1453,57 @@ def exported_candidates(
             continue
         source_id = _make_id(_file_stem(from_rel))
         target_id = _make_id(_file_stem(to_rel))
+
+        from_canon = _resolve_cached(from_path)
+        pkg_dir = to_path.parent
+        candidate_targets: set[str] = set()
+        try:
+            pkg_rel = pkg_dir.relative_to(root)
+            candidate_targets.add(_make_id(".".join(pkg_rel.parts)))
+            candidate_targets.add(_make_id(_file_stem(pkg_rel)))
+            candidate_targets.add(_make_id(str(pkg_rel)))
+        except ValueError:
+            pass
+        candidate_targets.add(_make_id(str(pkg_dir)))
+        candidate_targets.add(_make_id(str(pkg_dir / "__init__.py")))
+        candidate_targets.add(_make_id(str(pkg_dir.with_suffix(".py"))))
+        try:
+            from_rel_parent = from_path.parent.relative_to(root)
+            candidate_targets.add(_make_id(str(from_rel_parent / "__init__.py")))
+            candidate_targets.add(_make_id(str(from_rel_parent.with_suffix(".py"))))
+        except ValueError:
+            pass
+
+        loc_str = f"L{line}"
+        loc_candidates = [
+            e for e in provisional_by_loc.get((from_canon, loc_str), [])
+            if id(e) not in retracted_edge_ids
+        ]
+        matched_edge = None
+        for edge in loc_candidates:
+            if edge.get("target") in candidate_targets:
+                matched_edge = edge
+                break
+        if matched_edge is None and len(loc_candidates) == 1:
+            matched_edge = loc_candidates[0]
+
+        if matched_edge is not None:
+            retracted_edge_ids.add(id(matched_edge))
+            existing_edges.discard((
+                str(matched_edge.get("source")),
+                str(matched_edge.get("target")),
+                str(matched_edge.get("relation")),
+                str(matched_edge.get("context") or ""),
+            ))
+
         add_edge(
             source_id, target_id, "imports_from", "submodule_import", line, from_path,
             local_alias=local_name if local_name != to_path.stem else None,
         )
 
+    if retracted_edge_ids:
+        edges[:] = [e for e in edges if id(e) not in retracted_edge_ids]
+
     # #2262 producer guard: never emit a `calls` use-edge from a source id
     # that owns no node. All node appends (ensure_symbol_node, declarations,
     # namespace exports) happened above, so the owned set is complete here.

tests/test_python_import_resolution.py

@@ -150,3 +150,103 @@ def test_python_parameter_return_and_generic_contexts(tmp_path: Path):
     assert ("process()", "Payload", "parameter_type") in pairs
     assert ("process()", "Result", "return_type") in pairs
     assert ("process_many()", "Payload", "generic_arg") in pairs
+
+
+def test_issue_3777_package_module_collision_phantom_cycle_absent(tmp_path: Path):
+    from graphify.analyze import find_import_cycles
+    from graphify.build import build_from_json
+
+    nettacker_py = _write(
+        tmp_path / "nettacker.py",
+        "from nettacker.main import run\n\ndef cli():\n    run()\n",
+    )
+    init_py = _write(tmp_path / "nettacker/__init__.py", "")
+    main_py = _write(
+        tmp_path / "nettacker/main.py",
+        "from nettacker.core.app import Nettacker\n\ndef run():\n    return Nettacker()\n",
+    )
+    app_py = _write(
+        tmp_path / "nettacker/core/app.py",
+        "from nettacker import logger\n\nclass Nettacker:\n    def start(self):\n        logger.log_info('start')\n",
+    )
+    logger_py = _write(
+        tmp_path / "nettacker/logger.py",
+        "def log_info(msg):\n    print(msg)\n",
+    )
+
+    result = extract(
+        [nettacker_py, init_py, main_py, app_py, logger_py],
+        cache_root=tmp_path,
+        root=tmp_path,
+    )
+
+    app_file = _node_id(result, "app.py", "nettacker/core/app.py")
+    logger_file = _node_id(result, "logger.py", "nettacker/logger.py")
+    nettacker_file = _node_id(result, "nettacker.py", "nettacker.py")
+
+    assert _has_edge(result, app_file, logger_file, "imports_from")
+    assert not _has_edge(result, app_file, nettacker_file, "imports_from")
+
+    graph = build_from_json(result)
+    assert find_import_cycles(graph) == []
+
+
+def test_issue_3777_nested_module_package_collision_resolves_to_submodule(tmp_path: Path):
+    from graphify.analyze import find_import_cycles
+    from graphify.build import build_from_json
+
+    runner_py = _write(
+        tmp_path / "pkg/runner.py",
+        "from pkg.runner.step import run\n\ndef start():\n    run()\n",
+    )
+    init_py = _write(tmp_path / "pkg/runner/__init__.py", "")
+    step_py = _write(
+        tmp_path / "pkg/runner/step.py",
+        "from pkg.runner import helper\n\ndef run():\n    helper.work()\n",
+    )
+    helper_py = _write(
+        tmp_path / "pkg/runner/helper.py",
+        "def work():\n    pass\n",
+    )
+
+    result = extract(
+        [runner_py, init_py, step_py, helper_py],
+        cache_root=tmp_path,
+        root=tmp_path,
+    )
+
+    step_file = _node_id(result, "step.py", "pkg/runner/step.py")
+    helper_file = _node_id(result, "helper.py", "pkg/runner/helper.py")
+    runner_file = _node_id(result, "runner.py", "pkg/runner.py")
+
+    assert _has_edge(result, step_file, helper_file, "imports_from")
+    assert not _has_edge(result, step_file, runner_file, "imports_from")
+
+    graph = build_from_json(result)
+    assert find_import_cycles(graph) == []
+
+
+def test_issue_3777_namespace_package_submodule_import(tmp_path: Path):
+    sub = _write(tmp_path / "ns/sub.py", "def helper():\n    pass\n")
+    consumer = _write(tmp_path / "ns/consumer.py", "from ns import sub\n")
+
+    result = extract([sub, consumer], cache_root=tmp_path, root=tmp_path)
+
+    consumer_file = _node_id(result, "consumer.py", "ns/consumer.py")
+    sub_file = _node_id(result, "sub.py", "ns/sub.py")
+
+    assert _has_edge(result, consumer_file, sub_file, "imports_from")
+
+
+def test_issue_3777_standalone_module_import_unaffected(tmp_path: Path):
+    standalone = _write(tmp_path / "standalone.py", "def fn():\n    return 42\n")
+    consumer = _write(tmp_path / "consumer.py", "from standalone import fn\n")
+
+    result = extract([standalone, consumer], cache_root=tmp_path, root=tmp_path)
+
+    consumer_file = _node_id(result, "consumer.py", "consumer.py")
+    standalone_file = _node_id(result, "standalone.py", "standalone.py")
+    fn_symbol = _node_id(result, "fn()", "standalone.py")
+
+    assert _has_edge(result, consumer_file, standalone_file, "imports_from")
+    assert _has_edge(result, consumer_file, fn_symbol, "imports")