feat(extract): add Kubernetes/ArgoCD/Helm-values YAML extractor

graphify/detect.py

@@ -442,6 +442,14 @@ def classify_file(path: Path) -> FileType | None:
     from graphify.manifest_ingest import is_package_manifest_path
     if is_package_manifest_path(path):
         return FileType.CODE
+    # Kubernetes / ArgoCD / External-Secrets manifests are deterministic YAML, so
+    # they take the AST path too. Sniffed by CONTENT (apiVersion + kind), never by
+    # extension: .yaml stays a DOC extension so ordinary YAML (CI configs, docs
+    # data) keeps going to the LLM path. Runs after _is_sensitive() in detect(),
+    # so a secrets.yaml is dropped before it can reach this branch.
+    from graphify.extractors.k8s import is_k8s_manifest_path
+    if is_k8s_manifest_path(path):
+        return FileType.CODE
     # Compound extensions must be checked before simple suffix lookup
     if path.name.lower().endswith(".blade.php"):
         return FileType.CODE

graphify/extract.py

@@ -43,6 +43,7 @@
 from graphify.extractors.fortran import _cpp_preprocess, extract_fortran  # noqa: F401
 from graphify.extractors.go import extract_go  # noqa: F401
 from graphify.extractors.json_config import extract_json  # noqa: F401
+from graphify.extractors.k8s import extract_k8s  # noqa: F401
 from graphify.extractors.markdown import extract_markdown  # noqa: F401
 from graphify.extractors.pascal_forms import extract_delphi_form, extract_lazarus_form  # noqa: F401
 from graphify.extractors.powershell import extract_powershell, extract_powershell_manifest  # noqa: F401
@@ -3923,6 +3924,10 @@ def add_existing_edge(edge: dict) -> None:
     ".tf": extract_terraform,
     ".tfvars": extract_terraform,
     ".hcl": extract_terraform,
+    # Only reached for YAML that detect.classify_file sniffed as a k8s manifest
+    # (apiVersion + kind); ordinary .yaml stays a document on the LLM path.
+    ".yaml": extract_k8s,
+    ".yml": extract_k8s,
     ".dm": extract_dm,
     ".dme": extract_dm,
     ".dmi": extract_dmi,
@@ -3952,6 +3957,8 @@ def add_existing_edge(edge: dict) -> None:
     ".hcl": "terraform",
     ".dm": "dm",
     ".dme": "dm",
+    ".yaml": "k8s",
+    ".yml": "k8s",
 }
 
 

graphify/extractors/__init__.py

@@ -20,6 +20,7 @@
 from graphify.extractors.go import extract_go
 from graphify.extractors.json_config import extract_json
 from graphify.extractors.julia import extract_julia
+from graphify.extractors.k8s import extract_k8s
 from graphify.extractors.markdown import extract_markdown
 from graphify.extractors.objc import extract_objc
 from graphify.extractors.pascal import extract_pascal
@@ -48,6 +49,7 @@
     "go": extract_go,
     "json": extract_json,
     "julia": extract_julia,
+    "k8s": extract_k8s,
     "lazarus_form": extract_lazarus_form,
     "markdown": extract_markdown,
     "objc": extract_objc,

graphify/extractors/k8s.py

@@ -0,0 +1,284 @@
+"""Kubernetes / Helm-values / ArgoCD-GitOps YAML extractor.
+
+Covers the manifests that are VALID YAML: plain k8s objects, ArgoCD
+Application/ApplicationSet/AppProject, External Secrets, and Helm `values*.yaml`.
+Helm *templates* (charts/*/templates/*.yaml) are deliberately out of scope — Go
+template directives (`{{- if .Values.x }}`) make them unparseable as YAML, so
+they are left to a later pass rather than parsed lossily here.
+
+Known gaps (deliberate, so the graph does not overstate what it knows):
+
+- An ApplicationSet's git generator yields a `generates_from` edge to the glob
+  it scans (`gitops/apps/*.yaml`), NOT to the files that glob matches. Resolving
+  it would mean globbing the filesystem from inside a single-file extractor,
+  which breaks per-file caching and determinism. The glob edge still shows which
+  registry each environment reads — the fan-out is one `ls` away.
+- Registry-style YAML that carries neither `apiVersion` nor `kind` (an ArgoCD
+  fleet entry is often just `app: <name>`) is indistinguishable from arbitrary
+  config, so it stays on the document path.
+- Objects are addressed `<kind>.<name>` globally, so same-named objects in
+  different clusters or namespaces (one `AppProject.platform` per environment)
+  collapse into a single node.
+"""
+from __future__ import annotations
+
+import re
+from pathlib import Path
+
+from graphify.extractors.base import _make_id
+
+# A k8s/ArgoCD manifest is identified by content, not by extension: `.yaml` is a
+# DOC extension and must STAY one for ordinary YAML (CI configs, front-matter,
+# docs data), which belongs on the LLM path. Only files carrying both
+# `apiVersion:` and `kind:` at the start of a line are rerouted to the AST path.
+# Mirrors manifest_ingest.is_package_manifest_path, which routes package
+# manifests (apm.yml, pyproject.toml) to CODE for the same reason (#1377).
+_APIVERSION_RE = re.compile(r"^apiVersion:\s*\S", re.MULTILINE)
+_KIND_RE = re.compile(r"^kind:\s*\S", re.MULTILINE)
+
+# Only the head of the file is sniffed — detect() calls this for every YAML in
+# the corpus, so it must stay cheap. A manifest declares both keys in its first
+# document; 8 KiB clears even a heavily commented one.
+_SNIFF_BYTES = 8192
+
+
+def is_k8s_manifest_path(path: Path) -> bool:
+    """True if `path` is a YAML file whose content looks like a k8s manifest."""
+    if path.suffix.lower() not in (".yaml", ".yml"):
+        return False
+    try:
+        with path.open("r", encoding="utf-8", errors="replace") as fh:
+            head = fh.read(_SNIFF_BYTES)
+    except OSError:
+        return False
+    return bool(_APIVERSION_RE.search(head) and _KIND_RE.search(head))
+
+# Injected into every mapping by _LineLoader so nodes get a real source_location.
+# Skipped during the walk so it never looks like a reference field.
+_LINE_KEY = "__graphify_line__"
+
+# Scalar-valued reference fields: `key: <name>` -> an object of a fixed kind.
+# Bare `name:` is deliberately absent — it identifies the enclosing object
+# rather than referencing another one, and treating it as a ref would make
+# every manifest point at itself.
+_SCALAR_REFS: dict[str, tuple[str, str]] = {
+    "project": ("AppProject", "references"),
+    "serviceAccountName": ("ServiceAccount", "references"),
+    "secretName": ("Secret", "references"),
+    "claimName": ("PersistentVolumeClaim", "references"),
+    "ingressClassName": ("IngressClass", "references"),
+    "storageClassName": ("StorageClass", "references"),
+    "priorityClassName": ("PriorityClass", "references"),
+}
+
+# Mapping-valued reference fields: `key: {name: <n>}`, optionally carrying its
+# own `kind:` (secretStoreRef/storeRef pick ClusterSecretStore vs SecretStore
+# that way). A None default means "read the kind from the mapping itself".
+_MAPPING_REFS: dict[str, tuple[str | None, str]] = {
+    "secretRef": ("Secret", "references"),
+    "secretKeyRef": ("Secret", "references"),
+    "configMapRef": ("ConfigMap", "references"),
+    "configMapKeyRef": ("ConfigMap", "references"),
+    "serviceAccountRef": ("ServiceAccount", "references"),
+    "secretStoreRef": (None, "references"),
+    "storeRef": (None, "references"),
+}
+
+def _line_loader():
+    """SafeLoader that records each mapping's 1-based start line.
+
+    PyYAML drops position information during construction, so without this every
+    node would have to share the file's L1. Constructed lazily because pyyaml is
+    an optional extra.
+    """
+    import yaml
+
+    class _LineLoader(yaml.SafeLoader):
+        pass
+
+    def _construct_mapping(loader, node, deep=False):
+        mapping = yaml.SafeLoader.construct_mapping(loader, node, deep=deep)
+        mapping[_LINE_KEY] = node.start_mark.line + 1
+        return mapping
+
+    _LineLoader.add_constructor(
+        yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, _construct_mapping
+    )
+    return _LineLoader
+
+
+def extract_k8s(path: Path) -> dict:
+    """Extract Kubernetes/ArgoCD objects and the references between them.
+
+    Nodes: one per YAML document carrying `apiVersion` + `kind`, addressed
+    `<kind>.<metadata.name>` (e.g. `Application.web-api-dev`), plus the
+    file itself and referenced-but-undeclared objects (a `project: platform` in one
+    file resolves to the `AppProject.platform` declared in another).
+
+    Edges: `contains` (file -> object), `references` (the _SCALAR_REFS /
+    _MAPPING_REFS table), `in_namespace`, `deploys_to` (Application ->
+    destination namespace), `deploys_chart` (Application -> spec.source.path),
+    `generates_from` (ApplicationSet -> its git generator's files/directories
+    globs) and `produces` (ExternalSecret -> the Secret it materializes).
+
+    Node IDs are scoped GLOBALLY by `<kind>.<name>`, not per-file or per-
+    directory, because that is how Kubernetes itself resolves a reference: an
+    ApplicationSet's `project: platform` means the AppProject named `platform`,
+    wherever it is declared. Global scoping is what lets those cross-file edges
+    survive the per-file extraction merge. Known limitation: same-named objects
+    in different clusters or namespaces (three `AppProject.platform` files, one
+    per env) collapse into one node.
+
+    Helm templates are NOT handled — see the module docstring.
+    """
+    try:
+        import yaml
+    except ImportError:
+        return {"nodes": [], "edges": [], "error": "pyyaml not installed. Run: pip install 'graphifyy[k8s]'"}
+
+    try:
+        loader = _line_loader()
+        source = path.read_text(encoding="utf-8", errors="replace")
+        docs = list(yaml.load_all(source, Loader=loader))
+    except Exception as e:
+        # Helm templates and other non-YAML land here; an unparseable file
+        # yields nothing rather than aborting the whole extraction run.
+        return {"nodes": [], "edges": [], "error": str(e)}
+
+    str_path = str(path)
+    file_nid = _make_id(str_path)
+
+    nodes: list[dict] = [{"id": file_nid, "label": path.name, "file_type": "code",
+                          "source_file": str_path, "source_location": None}]
+    edges: list[dict] = []
+    seen_ids: set[str] = {file_nid}
+    seen_edges: set[tuple[str, str, str]] = set()
+
+    def _addr_id(address: str) -> str:
+        return _make_id("k8s", address)
+
+    def _add_node(address: str, line: int, *, declared: bool) -> str:
+        """Add (or reuse) a node for `address`. `declared` marks a real
+        definition in this file, which also earns a `contains` edge from it;
+        a reference-only target gets a node so the edge has somewhere to land,
+        and the file that declares it will fill in the location on merge."""
+        nid = _addr_id(address)
+        if nid not in seen_ids:
+            seen_ids.add(nid)
+            nodes.append({"id": nid, "label": address, "file_type": "code",
+                          "source_file": str_path if declared else "",
+                          "source_location": f"L{line}" if declared else None})
+        if declared:
+            _add_edge(file_nid, address, "contains", line)
+        return nid
+
+    def _add_edge(src: str, address: str, relation: str, line: int) -> None:
+        tgt = _addr_id(address)
+        if src == tgt:
+            return
+        key = (src, tgt, relation)
+        if key in seen_edges:
+            return
+        seen_edges.add(key)
+        edges.append({"source": src, "target": tgt, "relation": relation,
+                      "confidence": "EXTRACTED", "source_file": str_path,
+                      "source_location": f"L{line}", "weight": 1.0})
+
+    def _ref(owner: str, kind: str, name, relation: str, line: int) -> None:
+        if not isinstance(name, str) or not name.strip():
+            return
+        address = f"{kind}.{name.strip()}"
+        _add_node(address, line, declared=False)
+        _add_edge(owner, address, relation, line)
+
+    def _line_of(mapping, fallback: int) -> int:
+        if isinstance(mapping, dict):
+            got = mapping.get(_LINE_KEY)
+            if isinstance(got, int):
+                return got
+        return fallback
+
+    def _walk(value, owner: str, line: int) -> None:
+        """Recursively scan a document for reference fields.
+
+        Depth-agnostic on purpose: an ApplicationSet nests a whole Application
+        spec under `spec.template.spec`, so matching on key NAME rather than a
+        fixed path picks those up with no ApplicationSet-specific code (DRY).
+        """
+        if isinstance(value, list):
+            for item in value:
+                _walk(item, owner, line)
+            return
+        if not isinstance(value, dict):
+            return
+
+        here = _line_of(value, line)
+
+        for key, val in value.items():
+            if key == _LINE_KEY:
+                continue
+            kline = _line_of(val, here)
+
+            if key in _SCALAR_REFS and isinstance(val, str):
+                kind, relation = _SCALAR_REFS[key]
+                _ref(owner, kind, val, relation, kline)
+            elif key in _MAPPING_REFS and isinstance(val, dict):
+                kind, relation = _MAPPING_REFS[key]
+                _ref(owner, kind or val.get("kind", "Secret"), val.get("name"), relation, kline)
+            elif key == "destination" and isinstance(val, dict):
+                # ArgoCD Application/ApplicationSet target cluster+namespace.
+                _ref(owner, "Namespace", val.get("namespace"), "deploys_to", kline)
+            elif key in ("source", "sources") and isinstance(val, (dict, list)):
+                for src in (val if isinstance(val, list) else [val]):
+                    if isinstance(src, dict) and isinstance(src.get("path"), str):
+                        _path_ref(owner, src["path"], "deploys_chart", _line_of(src, kline))
+            elif key == "git" and isinstance(val, dict):
+                # ApplicationSet git generator — `files:`/`directories:` globs are
+                # the registry an AppSet fans out over. This is the edge that
+                # makes "which apps reach prod?" answerable.
+                for bucket in ("files", "directories"):
+                    for entry in val.get(bucket) or []:
+                        if isinstance(entry, dict) and isinstance(entry.get("path"), str):
+                            _path_ref(owner, entry["path"], "generates_from", _line_of(entry, kline))
+            elif key == "target" and isinstance(val, dict) and isinstance(val.get("name"), str):
+                # ExternalSecret materializes a Secret of this name.
+                _ref(owner, "Secret", val["name"], "produces", kline)
+
+            _walk(val, owner, kline)
+
+    def _path_ref(owner: str, raw: str, relation: str, line: int) -> None:
+        """Reference to a repo path (a chart dir, or a generator glob).
+
+        Kept as its own `path.<value>` address rather than resolved to a file
+        node: a generator path is usually a glob (`gitops/apps/*.yaml`) and a
+        chart path is often templated (`charts/{{.app}}`), so neither maps to a
+        single real file. Naming them keeps the relationship queryable.
+        """
+        raw = raw.strip()
+        if not raw:
+            return
+        address = f"path.{raw}"
+        _add_node(address, line, declared=False)
+        _add_edge(owner, address, relation, line)
+
+    for doc in docs:
+        if not isinstance(doc, dict):
+            continue
+        kind = doc.get("kind")
+        meta = doc.get("metadata")
+        if not isinstance(kind, str) or not kind.strip() or not doc.get("apiVersion"):
+            continue
+        name = meta.get("name") if isinstance(meta, dict) else None
+        if not isinstance(name, str) or not name.strip():
+            continue
+
+        line = _line_of(doc, 1)
+        address = f"{kind.strip()}.{name.strip()}"
+        owner = _add_node(address, line, declared=True)
+
+        if isinstance(meta, dict) and isinstance(meta.get("namespace"), str):
+            _ref(owner, "Namespace", meta["namespace"], "in_namespace", _line_of(meta, line))
+
+        _walk(doc.get("spec"), owner, line)
+
+    return {"nodes": nodes, "edges": edges}

pyproject.toml

@@ -70,6 +70,12 @@ gemini = ["openai", "tiktoken"]
 openai = ["openai", "tiktoken"]
 chinese = ["jieba"]
 sql = ["tree-sitter-sql"]
+# extract_k8s() parses Kubernetes/ArgoCD manifests and Helm values with PyYAML.
+# Optional (like sql/terraform above): without it those files are detected and
+# skipped with the standard missing-extra warning rather than failing the run.
+k8s = [
+    "pyyaml>=6.0",
+]
 # extract_pascal() uses tree-sitter-pascal for AST-quality extraction (more
 # accurate calls/inherits edges) and falls back to a regex extractor when it is
 # absent (#781), so this stays optional. Unlike tree-sitter-dm below, it ships
@@ -80,7 +86,7 @@ pascal = ["tree-sitter-pascal"]
 # avoids breaking the default `uv tool install graphifyy` for everyone (#1104).
 dm = ["tree-sitter-dm"]
 terraform = ["tree-sitter-hcl"]
-all = ["mcp", "starlette>=1.3.1", "neo4j", "falkordb", "pypdf>=6.12.0", "markdownify", "watchdog", "graspologic; python_version < '3.13'", "python-docx", "openpyxl", "faster-whisper; python_version >= '3.11'", "yt-dlp>=2026.6.9", "matplotlib", "numpy>=2.0; python_version >= '3.13'", "openai", "tiktoken", "boto3", "anthropic", "tree-sitter-sql", "jieba", "tree-sitter-dm", "tree-sitter-hcl", "tree-sitter-pascal"]
+all = ["mcp", "starlette>=1.3.1", "neo4j", "falkordb", "pypdf>=6.12.0", "markdownify", "watchdog", "graspologic; python_version < '3.13'", "python-docx", "openpyxl", "faster-whisper; python_version >= '3.11'", "yt-dlp>=2026.6.9", "matplotlib", "numpy>=2.0; python_version >= '3.13'", "openai", "tiktoken", "boto3", "anthropic", "tree-sitter-sql", "jieba", "tree-sitter-dm", "tree-sitter-hcl", "tree-sitter-pascal", "pyyaml>=6.0"]
 
 [project.scripts]
 graphify = "graphify.__main__:main"

tests/test_extractors_registry.py

@@ -42,3 +42,12 @@ def test_terraform_migrated():
 
     assert facade.extract_terraform is extract_terraform
     assert LANGUAGE_EXTRACTORS["terraform"] is extract_terraform
+
+
+def test_k8s_registered():
+    # extract_k8s is a NEW extractor (not a migration), but it obeys the same
+    # contract: one object, reachable identically via the facade and registry.
+    from graphify.extractors.k8s import extract_k8s
+
+    assert facade.extract_k8s is extract_k8s
+    assert LANGUAGE_EXTRACTORS["k8s"] is extract_k8s