# Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Static read-only inspection for agent workflows."""
import ast
import os
import re
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional
import nvflare
from nvflare.tool.agent import frameworks
from nvflare.tool.agent.frameworks.base import DetectContext
DEFAULT_MAX_FILES = 250
DEFAULT_MAX_FILE_BYTES = 512 * 1024
MAX_EVIDENCE_PER_BUCKET = 12
# After max_files is reached, the inspector accounts for a bounded number of
# unvisited files/directories so callers can see that classification is
# incomplete without turning the cap into an unbounded full-tree walk.
MAX_FILE_LIMIT_ACCOUNTED_SKIPS = 10000
# Backstop for evidence collected per framework bucket. Far above the display
# cap so ranking/detection uses true counts; only a memory guard for pathological
# inputs, not a routing-relevant threshold.
MAX_EVIDENCE_COLLECT = 10000
# Packaging root dirs whose leading segment is not part of the import path
# (PyPA src-layout), so `src/pkg/mod.py` is importable as `pkg.mod`.
_PACKAGE_ROOT_DIR_NAMES = {"src"}
PYTHON_SUFFIXES = {".py"}
SKIPPED_DIR_NAMES = {
".git",
".hg",
".mypy_cache",
".pytest_cache",
".ruff_cache",
".tox",
".venv",
"__pycache__",
"build",
"dist",
"env",
"node_modules",
"venv",
}
SENSITIVE_FILE_SUFFIXES = {".key", ".pem", ".p12", ".pfx"}
SENSITIVE_FILE_NAMES = {"id_rsa", "id_dsa", "id_ecdsa", "id_ed25519"}
SECRET_NAME_PATTERN = re.compile(r"(api[_-]?key|secret|token|password|passwd|credential|access[_-]?key)", re.I)
_INCOMPLETE_SCAN_SKIP_CODES = {
"DIRECTORY_NOT_SCANNED_FILE_LIMIT",
"FILE_LIMIT_REACHED",
"FILE_TOO_LARGE",
"NON_UTF8_FILE",
"UNREADABLE_DIRECTORY",
"UNREADABLE_FILE",
}
# Installed-skill discovery: read-only scan for <dir>/*/SKILL.md under known
# agent skill directories. Bounded so a pathological tree can't blow up the scan.
SKILL_FILE_NAME = "SKILL.md"
MAX_INSTALLED_SKILLS = 200
MAX_SKILL_FRONTMATTER_BYTES = 64 * 1024
# Project-scope skill dirs are resolved relative to the inspected path's project
# root (walked up to cwd); global-scope dirs live under the user home.
_PROJECT_SKILL_DIRS = (".claude/skills", ".agents/skills")
_GLOBAL_SKILL_DIRS = ("~/.claude/skills", "~/.codex/skills")
# Framework detection (import roots, symbols, evidence weights, recommended
# skills, and family/promotion rules) lives in nvflare.tool.agent.frameworks.
# This engine stays framework-agnostic; add a framework there, not here.
[docs]
@dataclass
class InspectState:
root: Path
redact: bool
entries_visited: int = 0
files_considered: int = 0
files_scanned: int = 0
bytes_scanned: int = 0
files_skipped_count: int = 0
file_limit_reached: bool = False
file_limit_accounted_skips: int = 0
file_limit_skip_accounting_truncated: bool = False
classification_incomplete: bool = False
files_skipped: list[dict] = field(default_factory=list)
findings: list[dict] = field(default_factory=list)
framework_evidence: dict[str, list[dict]] = field(default_factory=dict)
flare_imports: list[dict] = field(default_factory=list)
flare_calls: set[str] = field(default_factory=set)
# framework name -> FLARE conversion-integration call names (e.g. Lightning
# flare.patch). Populated by framework detectors; used by _conversion_state.
integration_signals: dict[str, set[str]] = field(default_factory=dict)
file_imports: dict[str, set[str]] = field(default_factory=dict)
entry_points: list[dict] = field(default_factory=list)
job_py: Optional[str] = None
sim_env_used: bool = False
export_support: bool = False
exported_job_markers: list[str] = field(default_factory=list)
exported_job_marker_paths: list[Path] = field(default_factory=list)
distributed_patterns: list[dict] = field(default_factory=list)
dynamic_patterns: list[dict] = field(default_factory=list)
absolute_path_findings: list[dict] = field(default_factory=list)
# file -> list of (start_line, end_line) for every class definition. Used to
# decide whether base-framework usage lives inside a superset model class
# body (e.g. torch calls inside a LightningModule) versus standalone.
class_body_ranges: dict[str, list[tuple[int, int]]] = field(default_factory=dict)
# Cache for _local_files_by_module: built once after the scan populates
# file_imports, reused across per-evidence entry-context reachability checks.
local_files_by_module_cache: Optional[dict[str, set[str]]] = field(default=None, repr=False, compare=False)
[docs]
def inspect_path(
path: Path | str,
*,
redact: bool = True,
max_files: int = DEFAULT_MAX_FILES,
max_file_bytes: int = DEFAULT_MAX_FILE_BYTES,
) -> dict:
"""Inspect a path without importing or executing user code."""
target = _normalized_inspect_target(path)
state = InspectState(root=target, redact=redact)
if not target.exists() and not target.is_symlink():
raise FileNotFoundError(f"inspect path does not exist: {path}")
if target.is_symlink():
_record_symlink_skip(target, state)
elif target.is_file():
_inspect_file(target, state, max_file_bytes)
else:
_inspect_dir(target, state, max_files=max_files, max_file_bytes=max_file_bytes)
exported_job_info = _exported_job_info(state)
ranked_frameworks = _rank_frameworks(state)
detected_framework = _detect_primary_framework(state, ranked_frameworks)
ranked_frameworks = _order_frameworks_for_display(ranked_frameworks, detected_framework)
conversion_state = _conversion_state(state, detected_framework, exported_job_info)
target_type = _target_type(target, state, detected_framework, conversion_state)
return {
"schema_version": "1",
"nvflare_version": nvflare.__version__,
"path": _inspected_target_path(target),
"target_type": target_type,
"static_only": True,
"redaction": "on" if redact else "off",
"limits": {
"max_files": max_files,
"max_file_bytes": max_file_bytes,
"max_evidence_per_bucket": MAX_EVIDENCE_PER_BUCKET,
},
"classification_incomplete": state.classification_incomplete,
"scan": {
"entries_visited": state.entries_visited,
"files_considered": state.files_considered,
"files_scanned": state.files_scanned,
"bytes_scanned": state.bytes_scanned,
"files_skipped_count": state.files_skipped_count,
"files_skipped_count_approximate": state.file_limit_skip_accounting_truncated,
"files_skipped_truncated": state.file_limit_reached,
"files_skipped_evidence_truncated": state.files_skipped_count > len(state.files_skipped),
"files_skipped": state.files_skipped,
},
"frameworks": ranked_frameworks,
"entry_points": state.entry_points[:MAX_EVIDENCE_PER_BUCKET],
"flare_integration": {
"present": bool(state.flare_imports or state.flare_calls),
"imports": state.flare_imports[:MAX_EVIDENCE_PER_BUCKET],
"calls": sorted(state.flare_calls),
},
"conversion_state": conversion_state,
"job": {
"job_py": state.job_py,
"sim_env_used": state.sim_env_used,
"export_support": state.export_support,
"exported_job_markers": state.exported_job_markers[:MAX_EVIDENCE_PER_BUCKET],
"exported_job_candidates": exported_job_info["submit_ready_candidates"][:MAX_EVIDENCE_PER_BUCKET],
"nested_candidates": exported_job_info["nested_candidates"][:MAX_EVIDENCE_PER_BUCKET],
},
"patterns": {
"distributed": state.distributed_patterns[:MAX_EVIDENCE_PER_BUCKET],
"dynamic": state.dynamic_patterns[:MAX_EVIDENCE_PER_BUCKET],
"absolute_data_paths": state.absolute_path_findings[:MAX_EVIDENCE_PER_BUCKET],
},
"findings": state.findings[:MAX_EVIDENCE_PER_BUCKET],
"skill_selection": _skill_selection(detected_framework, conversion_state, state),
"recommended_next_commands": _recommended_next_commands(detected_framework, conversion_state, state),
"installed_skills": _installed_skills(target),
}
def _inspect_dir(root: Path, state: InspectState, *, max_files: int, max_file_bytes: int) -> None:
stack = [root]
while stack:
directory = stack.pop()
try:
children = sorted(directory.iterdir(), key=lambda p: p.name)
except OSError as e:
_add_skip(state, _skip_entry(directory, state, "UNREADABLE_DIRECTORY", "could not read directory", e))
continue
for index, child in enumerate(children):
if child.is_symlink():
_record_symlink_skip(child, state)
continue
if child.is_dir():
if _should_skip_dir(child, root):
_add_skip(state, _skip_entry(child, state, "DIRECTORY_SKIPPED", "directory skipped"))
continue
stack.append(child)
continue
if state.entries_visited >= max_files:
_record_unvisited_due_to_file_limit(state, root, stack, children[index:])
return
state.entries_visited += 1
if not child.is_file():
continue
_inspect_file(child, state, max_file_bytes)
if state.entries_visited >= max_files:
_record_unvisited_due_to_file_limit(state, root, stack, children[index + 1 :])
return
def _record_unvisited_due_to_file_limit(
state: InspectState, root: Path, pending_stack: list[Path], remaining_children: list[Path]
) -> None:
directories = list(pending_stack)
limit_left_unvisited_entries = bool(directories)
for child in remaining_children:
try:
if child.is_symlink():
if not _record_symlink_skip_after_file_limit(child, state):
break
continue
if child.is_file():
limit_left_unvisited_entries = True
if not _add_file_limit_skip(state, child):
break
continue
if not child.is_dir() or _should_skip_dir(child, root):
continue
except OSError:
limit_left_unvisited_entries = True
if not _add_skip_after_file_limit(
state, _skip_entry(child, state, "UNREADABLE_FILE", "could not stat file")
):
break
continue
directories.append(child)
limit_left_unvisited_entries = True
if not limit_left_unvisited_entries:
return
state.file_limit_reached = True
state.classification_incomplete = True
seen = set()
for directory in directories:
key = str(directory)
if key in seen:
continue
seen.add(key)
if not _add_directory_not_scanned_due_to_file_limit(state, directory):
break
_record_unvisited_files_under_file_limit(directory, state, root)
if state.file_limit_skip_accounting_truncated:
break
def _record_unvisited_files_under_file_limit(directory: Path, state: InspectState, root: Path) -> None:
stack = [directory]
while stack:
current = stack.pop()
try:
children = sorted(current.iterdir(), key=lambda p: p.name)
except OSError as e:
if not _add_skip_after_file_limit(
state, _skip_entry(current, state, "UNREADABLE_DIRECTORY", "could not read directory", e)
):
return
continue
for child in children:
try:
if child.is_symlink():
if not _record_symlink_skip_after_file_limit(child, state):
return
elif child.is_dir():
if not _should_skip_dir(child, root):
stack.append(child)
elif child.is_file():
if not _add_file_limit_skip(state, child):
return
except OSError:
if not _add_skip_after_file_limit(
state, _skip_entry(child, state, "UNREADABLE_FILE", "could not stat file")
):
return
def _account_file_limit_skip(state: InspectState) -> bool:
if state.file_limit_accounted_skips >= MAX_FILE_LIMIT_ACCOUNTED_SKIPS:
state.file_limit_skip_accounting_truncated = True
return False
state.file_limit_accounted_skips += 1
return True
def _add_file_limit_skip(state: InspectState, path: Path) -> bool:
if not _account_file_limit_skip(state):
return False
state.files_considered += 1
_add_skip(state, _skip_entry(path, state, "FILE_LIMIT_REACHED", "file scan limit reached"))
return True
def _add_skip_after_file_limit(state: InspectState, entry: dict) -> bool:
if not _account_file_limit_skip(state):
return False
_add_skip(state, entry)
return True
def _add_directory_not_scanned_due_to_file_limit(state: InspectState, directory: Path) -> bool:
return _add_skip_after_file_limit(
state,
_skip_entry(
directory,
state,
"DIRECTORY_NOT_SCANNED_FILE_LIMIT",
"directory not scanned because file scan limit was reached",
),
)
def _inspect_file(path: Path, state: InspectState, max_file_bytes: int) -> None:
state.files_considered += 1
rel_path = _display_path(path, state.root, state.redact)
if _is_sensitive_file(path):
_add_skip(state, _skip_entry(path, state, "SENSITIVE_FILE_SKIPPED", "sensitive file skipped"))
return
if _is_exported_job_marker(path):
state.exported_job_markers.append(rel_path)
state.exported_job_marker_paths.append(path)
if path.suffix not in PYTHON_SUFFIXES:
return
try:
size = path.stat().st_size
except OSError as e:
_add_skip(state, _skip_entry(path, state, "UNREADABLE_FILE", "could not stat file", e))
return
if size > max_file_bytes:
_add_skip(state, _skip_entry(path, state, "FILE_TOO_LARGE", "file exceeds static inspection cap"))
return
try:
# utf-8-sig strips a leading BOM (Windows/Notepad-authored sources) that
# would otherwise reach ast.parse as U+FEFF and raise SyntaxError, losing
# all framework/entry-point evidence for the file. It decodes plain UTF-8
# identically, so NON_UTF8_FILE handling is unaffected.
text = path.read_text(encoding="utf-8-sig")
except UnicodeDecodeError:
_add_skip(state, _skip_entry(path, state, "NON_UTF8_FILE", "file is not UTF-8 text"))
return
except OSError as e:
_add_skip(state, _skip_entry(path, state, "UNREADABLE_FILE", "could not read file", e))
return
state.files_scanned += 1
state.bytes_scanned += size
if path.name == "job.py":
state.job_py = rel_path
try:
tree = ast.parse(text, filename=str(path))
except SyntaxError as e:
state.findings.append(
{
"code": "PYTHON_PARSE_ERROR",
"severity": "warning",
"file": rel_path,
"line": e.lineno,
"message": "Python file could not be parsed statically.",
}
)
return
visitor = _PythonInspector(path, rel_path, state)
visitor.visit(tree)
_add_entry_point(path, rel_path, tree, state)
class _PythonInspector(ast.NodeVisitor):
def __init__(self, path: Path, rel_path: str, state: InspectState):
self.path = path
self.rel_path = rel_path
self.state = state
self._detectors = frameworks.detectors()
self._detector_states = {detector.name: detector.new_file_state() for detector in self._detectors}
self._ctx = DetectContext(
self._emit_framework_evidence,
self.state.flare_calls.add,
self._add_integration_signal,
)
def _emit_framework_evidence(self, framework: str, kind: str, value: str, lineno) -> None:
_append_evidence(self.state.framework_evidence, framework, _evidence(self.rel_path, lineno, kind, value))
def _add_integration_signal(self, framework: str, name: str) -> None:
self.state.integration_signals.setdefault(framework, set()).add(name)
def visit_Import(self, node: ast.Import) -> None:
for alias in node.names:
self._record_import(alias.name, node.lineno)
for detector in self._detectors:
detector.on_import(alias, self._detector_states[detector.name], self._ctx)
self.generic_visit(node)
def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
module = node.module or ""
self._record_import(_resolve_import_from_module(self.rel_path, module, node.level), node.lineno)
self._record_import_from_modules(module, node.level, node.names)
for detector in self._detectors:
detector.on_import_from(module, node.names, self._detector_states[detector.name], self._ctx)
for alias in node.names:
if alias.name in {"FedJob", "FLModel", "SimEnv"}:
self.state.flare_imports.append(
_evidence(self.rel_path, node.lineno, "from_import", f"{module}.{alias.name}")
)
self.generic_visit(node)
def visit_ClassDef(self, node: ast.ClassDef) -> None:
end_lineno = getattr(node, "end_lineno", None) or node.lineno
self.state.class_body_ranges.setdefault(self.rel_path, []).append((node.lineno, end_lineno))
for base in node.bases:
base_name = _symbol_name(base)
if not base_name:
continue
for detector in self._detectors:
detector.on_class_base(base_name, node.lineno, self._detector_states[detector.name], self._ctx)
self.generic_visit(node)
def visit_Call(self, node: ast.Call) -> None:
call_name = _call_name(node.func)
if call_name:
self._record_call(call_name, node.lineno)
for detector in self._detectors:
detector.on_call(call_name, node.lineno, self._detector_states[detector.name], self._ctx)
self.generic_visit(node)
def visit_Assign(self, node: ast.Assign) -> None:
self._inspect_secret_assignment(node.targets, node.value, getattr(node, "lineno", None))
self.generic_visit(node)
def visit_AnnAssign(self, node: ast.AnnAssign) -> None:
self._inspect_secret_assignment([node.target], node.value, getattr(node, "lineno", None))
self.generic_visit(node)
def visit_Constant(self, node: ast.Constant) -> None:
if isinstance(node.value, str):
self._inspect_string_literal(node.value, getattr(node, "lineno", None))
self.generic_visit(node)
def _record_import(self, module: str, lineno: int) -> None:
if not module:
return
self.state.file_imports.setdefault(self.rel_path, set()).add(module)
framework = frameworks.framework_for_import(module)
if framework:
_append_evidence(
self.state.framework_evidence,
framework,
_evidence(self.rel_path, lineno, "import", module),
)
if module == "nvflare" or module.startswith("nvflare."):
self.state.flare_imports.append(_evidence(self.rel_path, lineno, "import", module))
if module in {"hydra", "omegaconf"} or module.startswith(("hydra.", "omegaconf.")):
self.state.dynamic_patterns.append(_evidence(self.rel_path, lineno, "dynamic_config", module))
if module == "torch.distributed" or module.startswith("torch.distributed."):
self.state.distributed_patterns.append(_evidence(self.rel_path, lineno, "distributed_import", module))
if module == "accelerate" or module.startswith("accelerate."):
self.state.distributed_patterns.append(_evidence(self.rel_path, lineno, "accelerate_import", module))
def _record_import_from_modules(self, module: str, level: int, aliases: list[ast.alias]) -> None:
resolved_module = _resolve_import_from_module(self.rel_path, module, level)
imports = self.state.file_imports.setdefault(self.rel_path, set())
if resolved_module:
imports.add(resolved_module)
for alias in aliases:
if alias.name == "*":
continue
imports.add(f"{resolved_module}.{alias.name}" if resolved_module else alias.name)
def _record_call(self, call_name: str, lineno: int) -> None:
# Generic FLARE / distributed / dynamic-dispatch signals only. Ranked
# framework activity (pytorch_call, lightning_trainer) and conversion
# signals (flare.patch) are recorded by framework detectors via on_call.
if call_name.startswith("flare.") or call_name.startswith("nvflare."):
self.state.flare_calls.add(call_name)
if call_name in {"FedJob", "FLModel", "SimEnv"}:
self.state.flare_calls.add(call_name)
if call_name == "SimEnv" or call_name.endswith(".SimEnv"):
self.state.sim_env_used = True
if call_name.endswith(".export"):
self.state.export_support = True
if call_name in {"importlib.import_module", "__import__", "getattr"}:
self.state.dynamic_patterns.append(_evidence(self.rel_path, lineno, "dynamic_dispatch", call_name))
if call_name == "torch.compile":
self.state.dynamic_patterns.append(_evidence(self.rel_path, lineno, "torch_compile", call_name))
if call_name.endswith(("DataParallel", "FSDP", "Accelerator")):
self.state.distributed_patterns.append(_evidence(self.rel_path, lineno, "distributed_call", call_name))
def _inspect_secret_assignment(self, targets: list[ast.AST], value: ast.AST, lineno: Optional[int]) -> None:
if not isinstance(value, ast.Constant) or not isinstance(value.value, str):
return
for target in targets:
name = _target_name(target)
if name and SECRET_NAME_PATTERN.search(name):
self.state.findings.append(
{
"code": "SECRET_LITERAL_REDACTED",
"severity": "warning",
"file": self.rel_path,
"line": lineno,
"name": name,
"value": "<REDACTED>" if self.state.redact else value.value,
}
)
def _inspect_string_literal(self, value: str, lineno: Optional[int]) -> None:
if _looks_like_absolute_path(value):
self.state.absolute_path_findings.append(
{
"code": "ABSOLUTE_DATA_PATH",
"severity": "warning",
"file": self.rel_path,
"line": lineno,
"pattern_type": "absolute_path_literal",
"value": _redact_literal(value, self.state.redact),
}
)
def _rank_frameworks(state: InspectState) -> list[dict]:
total = sum(len(evidence) for evidence in state.framework_evidence.values())
ranked = []
for framework, evidence in state.framework_evidence.items():
count = len(evidence)
confidence = 0.0
if total:
# Any static import evidence should register clearly, but cap below
# certainty because import presence alone does not prove active use.
confidence = min(0.99, 0.45 + (count / total) * 0.5)
ranked.append(
{
"name": framework,
"confidence": round(confidence, 2),
"evidence": evidence[:MAX_EVIDENCE_PER_BUCKET],
"contradicting_evidence": [],
}
)
return sorted(ranked, key=lambda item: (-item["confidence"], item["name"]))
def _detect_primary_framework(state: InspectState, ranked: list[dict]) -> Optional[str]:
if not ranked:
return None
primary = _primary_by_confidence_and_entry_context(state, ranked)
return frameworks.resolve_primary_framework(primary, state.framework_evidence, _FamilyResolver(state))
def _primary_by_confidence_and_entry_context(state: InspectState, ranked: list[dict]) -> str:
# Frameworks are ranked by (confidence, name), which is count-based and blind
# to reachability: an incidental torch/sklearn utility in an unreachable
# helper file can outrank the framework the entry point actually uses. When
# any framework's evidence is tied to the entry context (the inspected file
# or a file reachable from an entry point), prefer the highest-confidence
# such framework over a higher-count-but-unreachable one. Same-family
# (base/superset) promotion is resolved afterward by
# resolve_primary_framework. Fall back to the raw ranking only when nothing
# is reachable (e.g. a model-only directory with no entry point).
#
# A numerical utility (numpy) is skipped here: an incidental `import numpy`
# in the entry must not win over a real convertible framework whose code is
# loaded dynamically or lives outside the entry file. Utilities are still
# ranked and reported; they just never become the entry-context primary or
# fallback primary while a non-utility framework exists.
#
# Selection order:
# 1. Highest-confidence non-utility framework with ACTIVE evidence tied to
# the entry context (a real model/usage the entry actually reaches).
# 2. Else highest-confidence non-utility framework with ANY evidence tied
# to the entry context (an import-only entry framework such as sklearn
# when no active framework is reachable there).
# 3. Else highest-confidence non-utility framework; a utility wins only
# when it is the sole detected framework.
# Step 1 before step 2 means an actively-used torch model reachable from the
# entry outranks import-only entry evidence, without demoting a genuinely
# entry-owned import-only framework when nothing active is reachable.
#
# DESIGN DECISION (do not revert to a pure entry-context or pure count rule):
# this two-step order is the agreed reconciliation of two review positions.
# - When the entry reaches an ACTIVE convertible model (e.g. a torch
# nn.Module) alongside import-only sklearn, route to the active framework
# (recommend a conversion) rather than abstaining on the sklearn import.
# - When the torch model is NOT reachable from the entry (a stray helper),
# the entry-owned import-only framework still wins (sklearn -> abstain),
# preserving the earlier sklearn-entry fix. Both hold simultaneously.
for item in ranked:
name = item["name"]
if name in frameworks.UTILITY_FRAMEWORKS:
continue
active = [e for e in state.framework_evidence.get(name, []) if frameworks.is_active_evidence(name, e)]
if active and _framework_evidence_tied_to_entry_context(state, active):
return name
for item in ranked:
name = item["name"]
if name in frameworks.UTILITY_FRAMEWORKS:
continue
if _framework_evidence_tied_to_entry_context(state, state.framework_evidence.get(name, [])):
return name
for item in ranked:
if item["name"] not in frameworks.UTILITY_FRAMEWORKS:
return item["name"]
return ranked[0]["name"]
class _FamilyResolver:
"""Adapter giving family-owning detectors the engine's generic helpers.
A detector that resolves a base/superset family conflict (for example
Lightning over PyTorch) reads collected evidence, weighted scores, and
entry-context/import-graph checks through this adapter, so the engine holds
no framework-specific promotion logic.
"""
def __init__(self, state: InspectState):
self._state = state
def evidence(self, framework: str) -> list[dict]:
return self._state.framework_evidence.get(framework, [])
def active_evidence(self, framework: str) -> list[dict]:
return [item for item in self.evidence(framework) if frameworks.is_active_evidence(framework, item)]
def score(self, evidence: list[dict]) -> int:
return _evidence_score(evidence)
def tied_to_entry_context(self, evidence: list[dict]) -> bool:
return _framework_evidence_tied_to_entry_context(self._state, evidence)
def has_inspected_file_or_entry_point(self) -> bool:
return self._state.root.is_file() or bool(self._state.entry_points)
def evidence_outside_files(self, evidence: list[dict], reference_evidence: list[dict]) -> list[dict]:
reference_files = {item["file"] for item in reference_evidence}
return [item for item in evidence if item["file"] not in reference_files]
def evidence_outside_class_bodies(self, evidence: list[dict], class_evidence: list[dict]) -> list[dict]:
# Exclude items whose (file, line) falls within the body of a class named
# in ``class_evidence`` (matched by that class's definition line). Lets a
# family member (Lightning) claim base-framework (torch) calls inside its
# model class bodies without absorbing torch used in a sibling class or at
# module level in the same file.
ranges_by_file: dict[str, list[tuple[int, int]]] = {}
for item in class_evidence:
file_path = item["file"]
def_line = item.get("line")
for start, end in self._state.class_body_ranges.get(file_path, []):
if start == def_line:
ranges_by_file.setdefault(file_path, []).append((start, end))
return [
item for item in evidence if not _line_within_ranges(item.get("line"), ranges_by_file.get(item["file"]))
]
def _line_within_ranges(line: Optional[int], ranges: Optional[list[tuple[int, int]]]) -> bool:
if line is None or not ranges:
return False
return any(start <= line <= end for start, end in ranges)
def _framework_evidence_tied_to_entry_context(state: InspectState, evidence: list[dict]) -> bool:
if _framework_evidence_tied_to_inspected_file_or_entry_point(state, evidence):
return True
if state.root.is_file():
return False
return any(_entry_point_imports_file(state, item["file"]) for item in evidence)
def _framework_evidence_tied_to_inspected_file_or_entry_point(state: InspectState, evidence: list[dict]) -> bool:
if state.root.is_file():
inspected_file = _display_path(state.root, state.root, state.redact)
return any(item["file"] == inspected_file for item in evidence)
entry_point_paths = {entry["path"] for entry in state.entry_points}
return any(item["file"] in entry_point_paths for item in evidence)
def _entry_point_imports_file(state: InspectState, evidence_file: str) -> bool:
if not _module_names_for_file(evidence_file):
return False
local_files_by_module = _local_files_by_module(state)
for entry_point in state.entry_points:
if _imports_reach_file(
state,
entry_point["path"],
state.file_imports.get(entry_point["path"], set()),
evidence_file,
local_files_by_module,
):
return True
return False
def _imports_reach_file(
state: InspectState,
importing_file: str,
imports: set[str],
target_file: str,
local_files_by_module: dict[str, set[str]],
) -> bool:
# Match on the resolved file, not on module names: a stale src-layout copy
# (src/mypkg/loop.py) shares the stripped module name "mypkg.loop" with a
# root-level mypkg/loop.py, so name-based matching would let the entry point
# "reach" the never-imported copy. _local_files_by_module resolves the shared
# name to the root file only, and comparing files keeps them distinct.
pending_imports = [(importing_file, import_name) for import_name in imports]
seen_imports = set()
seen_files = set()
while pending_imports:
source_file, import_name = pending_imports.pop()
import_key = (source_file, import_name)
if import_key in seen_imports:
continue
seen_imports.add(import_key)
for imported_file in _local_files_for_import(import_name, source_file, local_files_by_module):
if imported_file in seen_files:
continue
seen_files.add(imported_file)
if imported_file == target_file:
return True
pending_imports.extend(
(imported_file, nested_import) for nested_import in state.file_imports.get(imported_file, set())
)
return False
def _local_files_by_module(state: InspectState) -> dict[str, set[str]]:
# Register every candidate module name for a file, including the src-layout
# root-stripped name (mypkg.loop from src/mypkg/loop.py). When a name is
# claimed by both a root-level file and a src/ copy, the collision is
# resolved per-import by _prefer_shared_packaging_root using the importing
# file's own packaging root, so neither a stale src/ copy nor a stale
# root-level copy can steal the actively-imported module in either direction.
#
# Memoized on the state: entry-context reachability calls this once per
# evidence item, and file_imports is fully populated during the scan before
# detection runs, so the map is built once instead of O(evidence) times over
# the now-uncapped evidence lists.
if state.local_files_by_module_cache is not None:
return state.local_files_by_module_cache
files_by_module: dict[str, set[str]] = {}
for file_path in state.file_imports:
for module_name in _module_names_for_file(file_path):
files_by_module.setdefault(module_name, set()).add(file_path)
state.local_files_by_module_cache = files_by_module
return files_by_module
def _packaging_root_of(file_path: str) -> str:
parts = Path(file_path).parts
if parts and parts[0] in _PACKAGE_ROOT_DIR_NAMES:
return parts[0]
return ""
def _prefer_shared_packaging_root(files: set[str], importing_file: str) -> set[str]:
# When an import resolves to copies in different packaging roots (a root-level
# file and a src/ copy of the same module path), prefer the copy sharing the
# importing file's packaging root. Fall back to all matches when none share
# it (e.g. a root-level entry importing a src-layout package).
if len(files) <= 1:
return files
importing_root = _packaging_root_of(importing_file)
same_root = {file_path for file_path in files if _packaging_root_of(file_path) == importing_root}
return same_root or files
def _local_files_for_import(
import_name: str, importing_file: str, local_files_by_module: dict[str, set[str]]
) -> set[str]:
files = set()
exact_candidates = _exact_module_candidates_for_import(import_name, importing_file, local_files_by_module)
resolved_modules = set()
for module_name in exact_candidates:
module_files = local_files_by_module.get(module_name, set())
if module_files:
resolved_modules.add(module_name)
files.update(module_files)
# Only follow a package's ``__init__.py`` once the full imported module path resolves to a
# local file. Otherwise an external absolute import (e.g. ``import lightning.pytorch``) whose
# leading segment happens to match an unrelated local package (a top-level ``lightning/``)
# would resolve that package's ``__init__.py`` and incorrectly promote it.
if not files:
return files
for module_name in _package_module_prefix_candidates_for_resolved(resolved_modules, exact_candidates):
files.update(
file_path
for file_path in local_files_by_module.get(module_name, set())
if _is_package_module_file(file_path)
)
return _prefer_shared_packaging_root(files, importing_file)
def _exact_module_candidates_for_import(
import_name: str, importing_file: str, local_files_by_module: dict[str, set[str]]
) -> set[str]:
candidates = {import_name} if import_name else set()
context_prefix = _import_context_prefix(importing_file)
if context_prefix:
for module_name in list(candidates):
prefixed = f"{context_prefix}.{module_name}"
# Single-segment imports always take the importing file's context prefix so a sibling
# module (``from block import ...`` next to the script) resolves. Dotted imports only
# take it when the full context-prefixed module resolves to a local file. This keeps
# nested local dotted imports reachable (``from layers.block import ...`` in
# ``models/train.py`` resolving ``models/layers/block.py``) without promoting an
# external absolute import (``import lightning.pytorch``) onto an unrelated local
# package.
if _is_single_segment_import(module_name) or prefixed in local_files_by_module:
candidates.add(prefixed)
return candidates
def _package_module_prefix_candidates_for_resolved(resolved_modules: set[str], exact_candidates: set[str]) -> set[str]:
# Package-prefix candidates let us follow the __init__.py of a package whose full path
# resolves locally (e.g. ``import pkg.sub`` reaching ``pkg/__init__.py``). Derive the prefixes
# from the exact module candidates that actually resolved to local files, not from the raw
# import name. A context-resolved import (``from layers.block import ...`` in ``models/train.py``
# resolving ``models.layers.block``) must follow ``models.layers`` prefixes, not the raw
# ``layers`` segment that could match an unrelated top-level ``layers/`` package. The exact
# candidates are excluded so we only follow parent packages, not the fully resolved module file.
candidates: set[str] = set()
for module_name in resolved_modules:
candidates.update(_module_name_prefixes(module_name))
candidates.difference_update(exact_candidates)
return candidates
def _is_single_segment_import(import_name: str) -> bool:
return bool(import_name) and "." not in import_name
def _module_name_prefixes(module_name: str) -> set[str]:
parts = [part for part in module_name.split(".") if part]
return {".".join(parts[:index]) for index in range(1, len(parts) + 1)}
def _is_package_module_file(file_path: str) -> bool:
return Path(file_path).name == "__init__.py"
def _file_module_parts(file_path: str) -> Optional[tuple[str, ...]]:
if not file_path.endswith(".py"):
return None
path = Path(file_path)
parts = path.parent.parts if path.name == "__init__.py" else path.with_suffix("").parts
if not parts or any(part in {"", ".", ".."} for part in parts):
return None
return parts
def _module_names_for_file(file_path: str) -> set[str]:
parts = _file_module_parts(file_path)
if not parts:
return set()
names = {".".join(parts)}
# src-layout: a file under a packaging root (src/) is imported by its
# package path without the root, so an entry point's `import mypkg.loop`
# reaches src/mypkg/loop.py. Offer the root-stripped module name too.
if len(parts) > 1 and parts[0] in _PACKAGE_ROOT_DIR_NAMES:
names.add(".".join(parts[1:]))
return names
def _import_context_prefix(file_path: str) -> str:
if not file_path.endswith(".py"):
return ""
path = Path(file_path)
parts = path.parent.parts
if any(part in {"", ".", ".."} for part in parts):
return ""
return ".".join(parts)
def _resolve_import_from_module(importing_file: str, module: str, level: int) -> str:
if level <= 0:
return module
# The same keep-formula is correct for both plain modules and __init__.py.
package_parts = Path(importing_file).parent.parts
keep = max(0, len(package_parts) - level + 1)
parts = list(package_parts[:keep])
if module:
parts.extend(module.split("."))
return ".".join(part for part in parts if part)
def _evidence_score(evidence: list[dict]) -> int:
weights = frameworks.evidence_weights()
return sum(weights.get(item["kind"], 1) for item in evidence)
def _order_frameworks_for_display(ranked: list[dict], detected_framework: Optional[str]) -> list[dict]:
# Surface the detected primary framework first so callers reading
# frameworks[0] always stay aligned with the routing decision, including the
# family case (a PyTorch base detected while a higher-confidence Lightning
# member is present). sorted() is stable, so every other framework keeps its
# confidence-ranked order. When nothing was detected, the order is unchanged.
if not detected_framework:
return ranked
return sorted(ranked, key=lambda item: item["name"] != detected_framework)
def _exported_job_info(state: InspectState) -> dict:
root = state.root if state.root.is_dir() and not state.root.is_symlink() else state.root.parent
markers_by_dir: dict[Path, set[str]] = {}
for path in state.exported_job_marker_paths:
markers_by_dir.setdefault(path.parent, set()).add(path.name)
valid_candidate_dirs = set()
consumed_marker_dirs = set()
for directory, names in markers_by_dir.items():
if "meta.json" in names and names.intersection({"config_fed_server.json", "config_fed_client.json"}):
valid_candidate_dirs.add(directory)
meta_dirs = {directory for directory, names in markers_by_dir.items() if "meta.json" in names}
config_paths = [
path
for path in state.exported_job_marker_paths
if path.name in {"config_fed_server.json", "config_fed_client.json"}
]
for meta_dir in meta_dirs:
for config_path in config_paths:
if config_path.parent.name == "config" and config_path.parent.parent.parent == meta_dir:
valid_candidate_dirs.add(meta_dir)
consumed_marker_dirs.add(config_path.parent)
submit_ready = sorted(
(_display_path(directory, root, state.redact) for directory in valid_candidate_dirs if directory == root)
)
nested = []
for directory, names in sorted(markers_by_dir.items(), key=lambda item: _display_path(item[0], root, state.redact)):
if directory in consumed_marker_dirs:
continue
rel_dir = _display_path(directory, root, state.redact)
if directory in valid_candidate_dirs:
if directory != root:
nested.append(
{
"path": rel_dir,
"markers": sorted(names),
"reason": "nested_exported_job_candidate",
}
)
else:
nested.append(
{
"path": rel_dir,
"markers": sorted(names),
"reason": "incomplete_exported_job_marker_set",
}
)
return {"submit_ready_candidates": submit_ready, "nested_candidates": nested}
def _conversion_state(state: InspectState, detected_framework: Optional[str], exported_job_info: dict) -> str:
if exported_job_info["submit_ready_candidates"]:
return "exported_job"
# job.py is a common filename (SLURM launchers) and SimEnv is a natural class
# name in RL/robotics code, so neither is trustworthy on its own. Require
# corroborating nvflare evidence (an nvflare-rooted import) before treating a
# name-only signal as a FLARE job, mirroring the exported-job marker grouping.
if (state.job_py or state.sim_env_used) and state.flare_imports:
return "flare_job"
if _has_conversion_integration(state):
return "client_api_converted"
if {"flare.receive", "flare.send"} <= state.flare_calls or "FLModel" in state.flare_calls:
return "client_api_converted"
if state.flare_imports or state.flare_calls:
return "partial_client_api"
if detected_framework:
return "not_converted"
return "unknown"
def _has_conversion_integration(state: InspectState) -> bool:
# A framework conversion-integration signal (e.g. an nvflare.client.lightning
# ``patch(trainer)`` call) is a definitive conversion signal even without an
# explicit ``flare.send``, because the framework's callback performs the
# result exchange. Detectors record these signals via on_call; do not
# require static constructor evidence here (wrappers/factories can hide it).
return bool(state.integration_signals and state.flare_imports)
def _target_type(path: Path, state: InspectState, detected_framework: Optional[str], conversion_state: str) -> str:
if path.is_symlink():
return "unknown_target"
if path.is_file():
return "single_training_script" if path.suffix == ".py" else "unknown_target"
if conversion_state == "exported_job":
return "exported_submit_ready_flare_job"
if conversion_state == "flare_job":
return "flare_job_source"
if detected_framework and conversion_state in {"partial_client_api", "client_api_converted"}:
return "mixed_workspace"
if frameworks.family_base_has_member(detected_framework, state.framework_evidence):
# A family base (e.g. PyTorch) detected alongside its superset member
# (PyTorch Lightning). Distinct from the FLARE conversion
# "mixed_workspace": two frameworks of the same family are present, not
# a partial FLARE conversion.
return "mixed_framework_workspace"
if detected_framework:
return "training_repository"
return "unknown_target"
def _add_entry_point(path: Path, rel_path: str, tree: ast.Module, state: InspectState) -> None:
functions = [node.name for node in tree.body if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))]
main_guard = any(_is_main_guard(node) for node in tree.body if isinstance(node, ast.If))
likely = path.name in {"client.py", "server.py", "train.py", "trainer.py", "main.py", "job.py"} or main_guard
if likely or any(name in {"main", "train", "fit", "evaluate"} for name in functions):
state.entry_points.append(
{
"path": rel_path,
"kind": "python_script",
"functions": functions[:MAX_EVIDENCE_PER_BUCKET],
"main_guard": main_guard,
}
)
def _is_main_guard(node: ast.If) -> bool:
left = getattr(node.test, "left", None)
comparators = getattr(node.test, "comparators", [])
if not isinstance(left, ast.Name) or left.id != "__name__" or not comparators:
return False
value = comparators[0]
return isinstance(value, ast.Constant) and value.value == "__main__"
def _installed_skills(target: Path) -> list[dict]:
"""Discover installed skills from known agent skill dirs (read-only).
Scans ``<dir>/*/SKILL.md`` under project-scope dirs (relative to the inspected
path's project root, walked up to cwd) and global-scope dirs (under the user
home). Reads only the YAML frontmatter ``name``/``description`` with a small
inline parser; no user code is imported or executed. Symlinked SKILL.md files
are skipped and results are deduplicated by skill name and capped.
"""
skills: list[dict] = []
seen_names: set[str] = set()
for base, scope in _installed_skill_search_roots(target):
for skill_dir in _iter_skill_dirs(base):
if len(skills) >= MAX_INSTALLED_SKILLS:
return skills
skill_file = skill_dir / SKILL_FILE_NAME
if skill_file.is_symlink() or not skill_file.is_file():
continue
frontmatter = _read_skill_frontmatter(skill_file)
if frontmatter is None:
continue
name = frontmatter.get("name") or skill_dir.name
if name in seen_names:
continue
seen_names.add(name)
skills.append(
{
"name": name,
"description": frontmatter.get("description", ""),
"scope": scope,
"source": _installed_skill_source(skill_dir),
}
)
return skills
def _installed_skill_search_roots(target: Path) -> list[tuple[Path, str]]:
roots: list[tuple[Path, str]] = []
project_root = _project_root_for(target)
if project_root is not None:
for rel in _PROJECT_SKILL_DIRS:
roots.append((project_root / rel, "project"))
home = Path.home()
for rel in _GLOBAL_SKILL_DIRS:
roots.append((Path(rel).expanduser() if rel.startswith("~") else home / rel, "global"))
return roots
def _project_root_for(target: Path) -> Optional[Path]:
# Walk up from the inspected path toward cwd looking for a directory that
# holds a known project-scope skill dir. Fall back to cwd so a project with
# no skills still reports an empty list rather than erroring.
try:
start = target if target.is_dir() and not target.is_symlink() else target.parent
start = start.resolve()
cwd = Path.cwd().resolve()
except OSError:
return None
candidates = [start, *start.parents]
for candidate in candidates:
for rel in _PROJECT_SKILL_DIRS:
if (candidate / rel).is_dir():
return candidate
if candidate == cwd:
break
return cwd
def _iter_skill_dirs(base: Path):
if base.is_symlink() or not base.is_dir():
return
try:
children = sorted(base.iterdir(), key=lambda p: p.name)
except OSError:
return
for child in children:
if child.is_symlink() or not child.is_dir():
continue
yield child
def _installed_skill_source(skill_dir: Path) -> str:
try:
return str(skill_dir.resolve(strict=False))
except OSError:
return str(skill_dir)
def _read_skill_frontmatter(skill_file: Path) -> Optional[dict]:
"""Parse the leading YAML frontmatter block for name/description only.
Small inline parser (no PyYAML, no dev-tools import): reads the block between
the leading ``---`` fences and extracts top-level ``name`` and ``description``
scalars. Returns None on unreadable/oversized files or a missing block.
"""
try:
if skill_file.stat().st_size > MAX_SKILL_FRONTMATTER_BYTES:
return None
text = skill_file.read_text(encoding="utf-8-sig", errors="replace")
except OSError:
return None
lines = text.splitlines()
if not lines or lines[0].strip() != "---":
return None
result: dict[str, str] = {}
for line in lines[1:]:
if line.strip() == "---":
break
# Only top-level keys (no leading indentation) so nested metadata is ignored.
if line[:1] in (" ", "\t") or ":" not in line:
continue
key, _, value = line.partition(":")
key = key.strip()
if key in ("name", "description"):
result[key] = _strip_scalar(value.strip())
return result
def _strip_scalar(value: str) -> str:
if len(value) >= 2 and value[0] == value[-1] and value[0] in ("'", '"'):
return value[1:-1]
return value
def _skill_selection(detected_framework: Optional[str], conversion_state: str, state: InspectState) -> dict:
recommended = []
if conversion_state == "exported_job":
# Lifecycle skills are out of scope and not planned; exported jobs are
# handled with product APIs directly, so no skill is recommended.
pass
elif detected_framework and conversion_state == "not_converted":
skill = frameworks.recommended_skill_for(detected_framework)
if skill:
recommended.append(skill)
if state.findings or _has_problematic_skips(state):
recommended.append("nvflare-orient")
return {
"detected_framework": detected_framework,
"conversion_state": conversion_state,
"exported_job": conversion_state == "exported_job",
"recommended_skills": recommended,
"safety_findings": [finding["code"] for finding in state.findings[:MAX_EVIDENCE_PER_BUCKET]],
}
def _has_problematic_skips(state: InspectState) -> bool:
return state.classification_incomplete
def _recommended_next_commands(
detected_framework: Optional[str], conversion_state: str, state: InspectState
) -> list[str]:
commands = []
if conversion_state == "exported_job":
commands.append("nvflare job submit <job-folder> --format json")
elif state.job_py and state.export_support and state.flare_imports:
# Only suggest `job.py --export` for a genuine FLARE job.py: `.export`
# calls (torch.onnx.export, YOLO model.export, ...) over-match, so without
# corroborating nvflare evidence this would ship a command that fails with
# an argparse error on an unrelated repo.
commands.append("python job.py --export --export-dir <job-dir>")
elif detected_framework and conversion_state == "not_converted":
skill = frameworks.recommended_skill_for(detected_framework)
if skill:
commands.append(f"Use the {skill} skill before editing.")
return commands
def _record_symlink_skip(path: Path, state: InspectState) -> None:
_add_skip(state, _symlink_skip_entry(path, state))
def _record_symlink_skip_after_file_limit(path: Path, state: InspectState) -> bool:
return _add_skip_after_file_limit(state, _symlink_skip_entry(path, state))
def _symlink_skip_entry(path: Path, state: InspectState) -> dict:
try:
target = os.readlink(path)
except OSError:
target = ""
return {
"code": "SYMLINK_SKIPPED",
"path": _display_path(path, state.root, state.redact),
"target": _redact_literal(target, state.redact),
"message": "symlink was not followed during static inspection",
}
def _add_skip(state: InspectState, entry: dict) -> None:
state.files_skipped_count += 1
if entry.get("code") in _INCOMPLETE_SCAN_SKIP_CODES:
state.classification_incomplete = True
if len(state.files_skipped) < MAX_EVIDENCE_PER_BUCKET:
state.files_skipped.append(entry)
def _skip_entry(path: Path, state: InspectState, code: str, message: str, error: Exception = None) -> dict:
result = {"code": code, "path": _display_path(path, state.root, state.redact), "message": message}
if error is not None:
result["error_type"] = type(error).__name__
return result
def _should_skip_dir(path: Path, root: Path) -> bool:
if path == root:
return False
return path.name in SKIPPED_DIR_NAMES or path.name.startswith(".")
def _is_sensitive_file(path: Path) -> bool:
return path.name in SENSITIVE_FILE_NAMES or path.suffix.lower() in SENSITIVE_FILE_SUFFIXES
def _is_exported_job_marker(path: Path) -> bool:
return path.name in {"meta.json", "config_fed_server.json", "config_fed_client.json"}
def _display_path(path: Path, root: Path, redact: bool) -> str:
base = root if root.is_dir() and not root.is_symlink() else root.parent
try:
return path.relative_to(base).as_posix()
except ValueError:
if redact and path.is_absolute():
return f"<REDACTED_PATH>/{path.name}"
return str(path)
def _inspected_target_path(path: Path) -> str:
return os.path.abspath(os.path.normpath(str(path)))
def _normalized_inspect_target(path: Path | str) -> Path:
return Path(_inspected_target_path(Path(path).expanduser()))
def _redact_literal(value: str, redact: bool) -> str:
if not redact:
return value
if _looks_like_absolute_path(value):
return "<REDACTED_PATH>"
if SECRET_NAME_PATTERN.search(value):
return "<REDACTED>"
return value
def _looks_like_absolute_path(value: str) -> bool:
return value.startswith(("/", "~")) or bool(re.match(r"^[A-Za-z]:[\\/]", value))
def _evidence(file_path: str, line: Optional[int], kind: str, value: str) -> dict:
return {"file": file_path, "line": line, "kind": kind, "value": value}
def _append_evidence(target: dict[str, list[dict]], key: str, value: dict) -> None:
# Collect up to a generous backstop so framework ranking/detection sees the
# true evidence counts. Display is truncated to MAX_EVIDENCE_PER_BUCKET
# separately (see _rank_frameworks); capping at collection time would skew the
# count-based confidence and let a file's first 12 imports decide routing.
bucket = target.setdefault(key, [])
if len(bucket) < MAX_EVIDENCE_COLLECT:
bucket.append(value)
def _call_name(node: ast.AST) -> Optional[str]:
if isinstance(node, ast.Name):
return node.id
if isinstance(node, ast.Attribute):
prefix = _call_name(node.value)
return f"{prefix}.{node.attr}" if prefix else node.attr
return None
def _symbol_name(node: ast.AST) -> Optional[str]:
if isinstance(node, ast.Subscript):
return _call_name(node.value)
return _call_name(node)
def _target_name(node: ast.AST) -> Optional[str]:
if isinstance(node, ast.Name):
return node.id
if isinstance(node, ast.Attribute):
return node.attr
return None