How State Transitions Work Between LangGraph Nodes in SkillSpector

State transitions in SkillSpector occur through a shared SkillspectorState TypedDict that LangGraph passes immutably between nodes, with each node returning a partial dictionary that gets merged into the evolving state.

SkillSpector, developed by NVIDIA, orchestrates its security analysis pipeline using LangGraph—a lightweight workflow engine that treats state transitions as functional reductions over a typed dictionary. The entire system relies on explicit state schema definitions and node contracts to ensure deterministic data flow from input resolution to final report generation.

The State Schema: SkillspectorState as the Shared Contract

The foundation of every state transition lives in src/skillspector/state.py, where the SkillspectorState TypedDict defines the contract shared across all nodes.

# Conceptual representation from src/skillspector/state.py

class SkillspectorState(TypedDict):
    skill_path: str
    components: list[Component]
    findings: Annotated[list[Finding], operator.add]  # Merge: concatenate lists

    filtered_findings: list[Finding]
    risk_score: int
    report_body: str
    # ... additional keys

Key design pattern: The findings field uses Annotated[list[Finding], operator.add] to instruct LangGraph to concatenate list outputs from multiple analyzers rather than overwrite them. Standard keys like skill_path or risk_score follow standard overwrite semantics—when a node returns a value for an existing key, the previous value is replaced.

Constructing the Workflow Graph

The graph topology is assembled in src/skillspector/graph.py using LangGraph’s StateGraph class. This file constructs the container and wires the state transitions:

from langgraph.graph import StateGraph, START, END
from skillspector.state import SkillspectorState

# Line ~30 in src/skillspector/graph.py

workflow = StateGraph(SkillspectorState)

# Nodes are registered with unique string IDs

workflow.add_node("resolve_input", resolve_input)
workflow.add_node("build_context", build_context)
workflow.add_node("meta_analyzer", meta_analyzer)
workflow.add_node("report", report)

# Dynamic analyzer nodes

for analyzer_id in ANALYZER_NODE_IDS:
    workflow.add_node(analyzer_id, ANALYZER_NODES[analyzer_id])

Edge declarations (lines 45–51 in graph.py) dictate the transition logic:

• START → resolve_input
• resolve_input → build_context
• build_context → <each analyzer> (fan-out to parallel analyzers)
• <each analyzer> → meta_analyzer (fan-in)
• meta_analyzer → report
• report → END

After wiring, the graph compiles to a callable object: graph = workflow.compile().

Node Execution and State Mutation

Each node in SkillSpector follows a strict contract: receive the current SkillspectorState, perform work, and return a partial dictionary containing only the keys being updated.

resolve_input Node

Located in src/skillspector/nodes/resolve_input.py, this entry point normalizes the user-provided input_path and returns:

return {
    "skill_path": resolved_absolute_path,
    "temp_dir_for_cleanup": temp_dir  # Optional

}

build_context Node

Implemented in src/skillspector/nodes/build_context.py, this node walks the skill directory and populates component metadata:

return {
    "components": discovered_components,
    "file_cache": file_contents_dict,
    "manifest": skill_manifest,
    "component_metadata": metadata,
    "has_executable_scripts": bool(executables)
}

Analyzer Nodes

The dynamic analyzer set (located in src/skillspector/nodes/analyzers/*.py) each read the file_cache and manifest, then return findings that append to the shared list:

# Inside any analyzer node

new_findings = [Finding(...), Finding(...)]
return {"findings": new_findings}  # Concatenated via Annotated operator.add

Because findings uses operator.add in the state schema, LangGraph automatically concatenates these lists as state transitions occur, preserving findings from previous analyzers.

meta_analyzer Node

Found in src/skillspector/nodes/meta_analyzer.py, this node consumes the accumulated findings, optionally invokes an LLM for enrichment, and writes a filtered view:

return {
    "filtered_findings": enriched_findings,
    "llm_metadata": metadata  # Optional tracking

}

report Node

The final node in src/skillspector/nodes/report.py computes risk scores and generates output:

return {
    "report_body": rendered_markdown,
    "sarif_report": sarif_dict,
    "risk_score": calculated_score,
    "risk_severity": severity_level
}

Runtime State Flow

At runtime, the compiled graph executes as a pure function. The state is immutable between edges—LangGraph copies the dictionary, applies the node’s returned updates, and passes the new dictionary forward. This guarantees deterministic replay and clear transition boundaries.

Execution sequence:

1. Start → resolve_input: Adds skill_path to the state.
2. resolve_input → build_context: Expands the directory structure, filling components and file_cache.
3. build_context → analyzers: Each analyzer reads the cache, appends Finding objects to findings.
4. All analyzers → meta_analyzer: Reads accumulated findings, writes filtered_findings.
5. meta_analyzer → report: Computes final risk metrics and generates report_body.
6. report → END: Returns the terminal state.

Invocation example:

from skillspector.graph import graph

initial_state = {
    "input_path": "https://github.com/example/my-skill.zip",
    "mode": "scan",
    "output_format": "json",
    "use_llm": True,
}

final_state = graph.invoke(initial_state)
print(final_state["risk_score"])      # 0-100 integer

print(final_state["report_body"])     # Rendered output

The graph.invoke() call triggers the entire pipeline, with LangGraph managing the state transition between nodes automatically.

Summary

• State contract: SkillspectorState in src/skillspector/state.py defines the TypedDict schema using Annotated types for custom merge behaviors.
• Graph construction: src/skillspector/graph.py wires nodes and edges, compiling to an executable graph object.
• Node semantics: Each node returns partial dictionaries; findings uses operator.add for list concatenation while other keys overwrite.
• Immutability: LangGraph copies state between transitions, ensuring deterministic, replayable execution.
• Entry point: Call graph.invoke(initial_state) to execute the full pipeline from input resolution to final report.

Frequently Asked Questions

How does SkillSpector handle concurrent analyzer writes to the state?

SkillSpector uses the Annotated[list[Finding], operator.add] type hint in src/skillspector/state.py to specify that the findings key should concatenate lists rather than overwrite them. When multiple analyzer nodes run and each returns {"findings": [finding1, finding2]}, LangGraph merges these into a single accumulated list using the addition operator, preserving all findings across parallel execution paths.

What happens if two nodes return values for the same state key?

Standard keys without special annotations follow overwrite semantics. If node_a returns {"risk_score": 50} and node_b later returns {"risk_score": 75}, the final state retains 75. Only keys explicitly annotated with merge operators (like the operator.add used for findings) receive special treatment during state transitions.

Can I execute the SkillSpector graph with partial state for debugging?

Yes. The compiled graph object in src/skillspector/graph.py accepts any dictionary matching the SkillspectorState schema. You can invoke intermediate entry points by providing pre-populated state (e.g., skipping resolve_input by passing {"skill_path": "/path/to/skill", "components": [...]}), though this requires manually satisfying the dependencies that downstream nodes expect.

Where is the state transition logic defined if I need to modify the workflow?

State transitions are defined by the edge declarations in src/skillspector/graph.py (lines 45–51). To change the flow, modify the workflow.add_edge() calls or use conditional edges with workflow.add_conditional_edges(). The state schema itself is controlled in src/skillspector/state.py, where you can add new keys or change merge behaviors using Annotated types.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}