# Dvp Entrypoint Identifier

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

- Skill: `snowflake-labs/dvp-entrypoint-identifier` (Agent Skill, multi-file: 4 files)
- Install (CLI): `npx skillmds@latest add snowflake-labs/dvp-entrypoint-identifier`
- Raw SKILL.md: https://api.skillmd.com/api/skills/snowflake-labs/dvp-entrypoint-identifier/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Integrations & APIs
- Author: snowflake-labs (https://skillmd.com/u/snowflake-labs)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/snowflake-labs/dvp-entrypoint-identifier

---


# 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.

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

### Step 1: Validate ASG Exists

**Before any processing**, verify the ASG file exists:

```python
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:

```bash
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.

```json
{
  "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:

```python
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:
```python
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](../docs/entrypoints-source-spec.md) |

Example:

```json
[
  {
    "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