dvp-entrypoint-identifier

name	dvp-entrypoint-identifier
description	Generate dvp/04-results/entrypoints.json from an ASG (Warp EntrypointDetector). Triggers: entrypoints, pipelines, entry point inventory.

DVP Entrypoint Identifier Skill (hybrid: deterministic + non-deterministic IA patch)

Overview

Hybrid approach:

• Step 1 (deterministic): generate a baseline entrypoints.json from the ASG using the embedded WARP EntrypointDetector.
• Step 2 (non-deterministic IA): evaluate the Step 1 output (including the case where it is empty or incomplete) and patch it by adding/adjusting/disabling entrypoints as needed.

Generates dvp/04-results/entrypoints.json from an ASG JSON file (dvp/04-results/*_asg.json) using the embedded WARP EntrypointDetector.

Goal

1. Run the deterministic embedded WARP EntrypointDetector using the ASG JSON to produce a baseline entrypoints.json.
2. Use non-deterministic IA to evaluate that baseline (including empty/incomplete output) and patch it by adding/adjusting/disabling entrypoints as required.

Inputs

Input	Required	Location
ASG JSON	MANDATORY	dvp/04-results/*_asg.json — generated by dvp-asg-generation. If this file does not exist, STOP with error. Do NOT attempt to generate entrypoints without the ASG.

Outputs

Output	Format	Location
Entrypoints inventory	JSON	dvp/04-results/entrypoints.json

Output Format

Every time you begin a step, sub-step, or significant action, prefix the message with a timestamp in the format [YYYY-MM-DD HH:MM:SS]. Obtain the current time by running date '+%Y-%m-%d %H:%M:%S' in bash.

Example:

[2026-03-24 14:05:32] Starting Step 1: Scan source files...
[2026-03-24 14:05:45] Detected 3 entrypoints
[2026-03-24 14:05:46] Step 1 complete.

Workflow

Step 0: Initialize Git

Ensure the workload directory has a git repository on the sma/migration-process branch. This is idempotent — if the orchestrator already initialized git, this is a no-op.

result = sma_api.git_ensure_ready("<workload_path>")

Step 1: Validate ASG Exists

Before any processing, verify the ASG file exists:

from pathlib import Path
asg_matches = sorted(Path("dvp/04-results").glob("*_asg.json"))
if not asg_matches:
    # ⛔ HARD STOP — do NOT fall back to AI-based source code analysis
    raise SystemExit(
        "ERROR: No ASG found in dvp/04-results/ (expected *_asg.json).\n"
        "The ASG is MANDATORY. Run dvp-asg-generation first.\n"
        "Do NOT attempt to identify entrypoints without the ASG."
    )

If no *_asg.json exists, STOP with error. Do NOT attempt to read source code directly as a fallback.

Step 2: Run deterministic logic

Run from repo root:

PYTHONPATH="skills/spark-migration/snowpark-api/dvp/lib:skills/spark-migration/snowpark-api/dvp/dvp-entrypoint-identifier/lib" python - <<'PY'
from pathlib import Path
import json

from entrypoints import EntrypointDetector

results_dir = Path("dvp/04-results")
asg_matches = sorted(results_dir.glob("*_asg.json"))
if not asg_matches:
    raise SystemExit(
        "No ASG found in dvp/04-results (expected *_asg.json). Run dvp-asg-generation first."
    )
if len(asg_matches) > 1:
    raise SystemExit(
        "Multiple ASG files found in dvp/04-results. Please choose the one to use:\n"
        + "\n".join(f"- {p}" for p in asg_matches)
    )

asg_path = asg_matches[0]

detector = EntrypointDetector()
detector.detect_from_file(asg_path)
entrypoints = detector.to_list()

out_path = results_dir / "entrypoints.json"
out_path.parent.mkdir(parents=True, exist_ok=True)
out_path.write_text(json.dumps(entrypoints, indent=2), encoding="utf-8")
print(f"Saved {len(entrypoints)} entry points to {out_path}")
PY

This produces one entrypoints.json. (Canonical results directory is dvp/04-results/.)

Step 3: IA validation + update entrypoints.json if required

1. Read dvp/04-results/entrypoints.json.
2. If it contains 0 active entrypoints:
   • Inspect the ASG and the workload source to add the most likely entrypoints.
3. Otherwise:
   • Validate a smart initial sample (cover multiple type, prioritize low confidence).
   • If incorrect entries are found, expand the sample; if errors persist above a reasonable threshold, validate all.
   • Update entrypoints.json:
     • Disable wrong entrypoints (status = disabled, add reason).
     • Add newly inferred entrypoints with origin = IA and status = detected.
     • Update entrypoints that need adjustments to be valid.

{
  "name": "workload",
  "source": "workload.py:134",
  "type": "script",
  "status": "disabled",
  "origin": "ASG",
  "reason": "..."
}

Rules:

• status: detected or disabled.
• origin: required. ASG for deterministic results, IA for AI-inferred entrypoints.
• reason: short and concrete explanation for why the entrypoint was disabled/added/updated.

This updates entrypoints.json by:

• disabling entries (status change + reason)
• adding IA entries (origin=IA + reason)

Stopping points

• If the patch would disable more than 30% of active entries, ask the user before applying (recommended).
• If there are no entrypoints after the IA validation and changes applied to entrypoints.json file.

Step 4: Commit Changes to Git

After entrypoints are identified and validated, commit the changes:

result = sma_api.git_commit("<workload_path>", """DVP Entrypoint Identifier: Identified N entrypoints

Active: N (ASG: X, IA: Y)
Disabled: M
Output: dvp/04-results/entrypoints.json""")

Verify branches:

result = sma_api.git_verify_branches("<workload_path>")

Notes

• The entrypoints are required, because test will be using them to create a unit test for each entrypoint.

Output schema

The output is a JSON array of objects with these fields:

Field	Type	Notes
name	string	Entrypoint name (usually file stem)
source	string	Hybrid locator <path>:<lineno>(::segment)*. Last :: segment is the method; preceding segments are scope. Python: workload.py:134, Scala: App.scala:5::MyApp::main, Notebook: notebook.py:1
type	string	script, module, or databricks_notebook
origin	string	Required: ASG or IA
status	string	Required: detected or disabled
reason	string	Detection reason (main_guard, notebook, spark_session_creation, main_method) or disable justification
inputs	object	{ total, by_type } rollup across transitive deps
outputs	object	{ total, by_type } rollup across transitive deps
adapted_source	string	(set by dvp-code-adapter, not by this skill) Post-adaptation invocation target, same hybrid format as source. See entrypoints-source-spec.md

Example:

[
  {
    "name": "workload",
    "origin": "ASG",
    "status": "detected",
    "source": "workload.py:134",
    "type": "script",
    "reason": "main_guard",
    "inputs": { "total": 3, "by_type": { "csv": 1, "parquet": 2 } },
    "outputs": { "total": 1, "by_type": { "delta": 1 } 
  

Final Summary

MANDATORY: After completing all steps (whether running standalone or invoked from the orchestrator), ALWAYS present this summary table:

Entrypoint Detection Complete

┌────────────────────────┬──────────┬──────────────────────────────────────────────┐
│ Step                   │ Status   │ Details                                      │
├────────────────────────┼──────────┼──────────────────────────────────────────────┤
│ Entrypoint Detection   │ Done     │ Identified N entrypoints                     │
└────────────────────────┴──────────┴──────────────────────────────────────────────┘

Output location: <output>/

Git branches:
• main — original code (unmodified)
• sma/migration-process — entrypoint detection changes applied

Rules:

• Replace N with actual count of entrypoints detected
• Status is Done, Skipped, or Failed
• If no entrypoints were found, show Skipped with reason
• If detection failed, show Failed with brief error
• The git branches section uses sma_api.git_verify_branches() to confirm both branches exist