Migrate Pyspark To Snowpark Connect

Migrate PySpark to SCOS — Coordinator

Orchestrate a multi-phase migration of PySpark workloads to Snowflake SCOS (Snowpark Connect for Spark). This coordinator delegates code-fixing (Phase 2) to a parallel pool of specialist sub-agents; runs the mechanical phases as scripts directly; and runs analysis + report-rendering (Phases 1, 1.2, 4) inline for single-file workloads or as sub-agents for multi-file workloads (sized by coordinator_mode). Every phase is validated with deterministic quality gates (scripts/scos_gates.py) before advancing.

When to Load

[snowpark-connect] Intent Detection: After user indicates migration intent (convert, migrate, update imports, rewrite for SCOS).

Arguments

• $ARGUMENTS — Path to the PySpark file or directory to migrate

Optional Metadata (from orchestrator)

Prerequisites

Skill Directory

<SKILL_DIRECTORY> is the parent snowpark-connect/ directory containing pyproject.toml and scripts/. All tool invocations use uv run --project <SKILL_DIRECTORY>.

uv Package Manager

Install uv if it is not already available. Show both OS variants — the skill must work on macOS / Linux and Windows:

# macOS / Linux
uv --version || curl -LsSf https://astral.sh/uv/install.sh | sh

# Windows (PowerShell)
uv --version; if ($LASTEXITCODE -ne 0) { irm https://astral.sh/uv/install.ps1 | iex }

The state model

A migration is a long, multi-phase run: nothing durable lives in this conversation, because it gets summarized or truncated before the run ends. Nor does it live in a checklist you keep in your head or in prose you write as you go — those drift out of step with what actually happened on disk. State lives in one structured, durable place: <CONVERSION>/migration_state.json.

phases_completed is the progress — it is not a log of it. The phase you are on is simply the first row of the phase table whose state key is absent; the phases that passed are the keys that are present. Read progress from the file rather than tracking it separately, and never let a phase's real outcome and its recorded outcome diverge — a phase that did not genuinely run must not be recorded as passed (see Phase Completeness).

After any compaction or context loss, rebuild from migration_state.json — never from memory. It carries the manifest, the paths, coordinator_mode, phases_completed, and per-file pending_files / processed_files progress: enough to resume exactly where the run stopped (see Resumption). You are also its single writer — specialists return results for you to merge, so the file stays consistent (see Universal Gate Contract).

Workflow

You are a coordinator. You NEVER hand-edit code fixes yourself, and the migrated reports are always produced by their generator scripts (never hand-written). Code fixing (Phase 2) is always delegated to specialist sub-agents via the task() tool — it fans out across files in a parallel worker pool and is by far the largest consumer of context, so isolating each fixer in its own sub-agent keeps the coordinator's window lean and lets waves run concurrently.

The analysis (Phase 1) and report-rendering (Phases 1.2 and 4) phases are run one of two ways depending on workload size, controlled by coordinator_mode (set in Phase 0: false when manifest holds a single file, true for multi-file workloads):

• coordinator_mode == false (single-file / small) → run them inline yourself, no sub-agent. Each is a deterministic script wrapped in at most light bounded judgment (a blind-spot supplementary scan for Phase 1, four advisory narrative sentences for Phase 1.2, nothing for Phase 4), and on a one-file workload the inlined reads are tiny — a sub-agent would buy only cold-start latency and context churn. For the Phase 1 supplementary scan, prefer grep/Bash over Read so the source stays out of your window.
• coordinator_mode == true (multi-file) → spawn a task() sub-agent for each, exactly as for the fixer. Here the supplementary scan reads many source files and analysis.json / AssessmentIR.json grow with the workload; doing that inline would accumulate in the coordinator's single session and push toward context_budget_tokens. The throwaway sub-agent absorbs those reads and returns only a compact summary, preserving context isolation.

In both cases agents/analyzer.md and agents/reporter.md are the canonical step-by-step procedures — read and follow them (inline, or as the sub-agent's prompt context); the procedure is identical, only who runs it changes. The mechanical phases (pre-processing, SQL rewrite, import/header updates, coverage, verification, validation) are always deterministic scripts you invoke directly. Every phase is validated with a deterministic quality gate (scripts/scos_gates.py or a phase-specific validator); the gates are the real enforcement and run identically regardless of who performed the phase.

Progress UI Emit Hooks (Experimental)

These one-liners feed the live dashboard. All are non-fatal (|| true). <SKILL_DIRECTORY> is the parent snowpark-connect/ directory. <CONVERSION> is the resolved Conversion-SCOS-<TIMESTAMP> path.

BUS="python3 <SKILL_DIRECTORY>/../scripts/progress_bus.py"
RUN="<CONVERSION>"
CONNECTION_NAME="<the Snowflake connection resolved in Phase 0>"

⚠️ EXPERIMENTAL — the Progress UI is OFF by default. Never launch it automatically. Start the dashboard only when the user has explicitly asked to see the Progress UI / live dashboard (e.g. "show the progress UI", "open the migration dashboard"). If they have not explicitly asked, skip the launch block below entirely — do not start the server and do not open a browser window. There must be no accidental launches. The event-emit commands elsewhere in this section are harmless no-ops when the server is not running, so continue to run them normally regardless.

Only if the user explicitly requested the Progress UI, launch the server once, right after $CONVERSION is resolved (this is the only place it is ever started; the top-level router and the SMA / snowpark-api path never launch it). --run "$RUN" must be the resolved <CONVERSION> dir so the server tails the same event log these hooks write to. The launcher requires the --confirm-experimental interlock (it is a no-op without it), is idempotent (reuses a live server), and is non-fatal (|| true) — migration never blocks on it.

UI="python3 <SKILL_DIRECTORY>/../scripts/ui_launch.py"
$UI start --run "$RUN" --skill-dir "<SKILL_DIRECTORY>/.." --port auto \
  --confirm-experimental \
  $([ "<config.ui_auto_open_browser>" = "yes" ] || echo "--no-browser") \
  2>/dev/null || true

Emit at each key boundary:

Phase names to use: assessment, preprocessing, sql-rewrite, analysis, report-assessment, migration, imports-headers, reports, verification, validation, notebook-migration.

Emit run-init once when $CONVERSION is first resolved (end of Phase 0 step 2), immediately after launching the server above, using the manifest length as total_files and the Phase-0 connection as connection_name. Because the server is started on $RUN (the resolved <CONVERSION> dir), no temp-sentinel re-init is needed.

The dashboard also includes an "Ask" chat for stakeholders to ask about live progress or the SCOS process. It is answered by SNOWFLAKE.CORTEX.COMPLETE over the same Snowflake connection this run uses — the server reads connection_name from the run-init event, so including it above is what keeps chat on the Cortex-enabled connection (the Phase-0 Cortex preflight already guarantees access) rather than falling back to default. Chat is read-only, no extra emissions required. If the connection can't be opened the chat shows a friendly error and the migration is unaffected.

Report linking is automatic. The server also scans the Reports/ folder and links MigrationReadinessReport.html, AssessmentIR.json, and the dashboard CSVs the moment they exist on disk, so they appear in the dashboard even if a report-ready hook is missed. The report-ready emits above are a belt-and-suspenders signal, not the only path.

Startup — Resume from Existing Assessment

Before reading any phase playbook, probe the output root for a completed assessment. This is deterministic: no handoff parameter is needed — the coordinator discovers state from disk.

Derive <OUTPUT_ROOT> as $OUTPUT if supplied, otherwise ${ARGUMENTS}_scos. Then run:

python3 -c "
import json, os, glob, sys
root = sys.argv[1]
found = []
for p in glob.glob(os.path.join(root, 'Conversion-SCOS-*', 'migration_state.json')):
    try:
        with open(p) as f:
            s = json.load(f)
        status = s.get('phases_completed', {}).get('1a_assessment_report', {}).get('status')
        if status == 'passed':
            found.append(os.path.dirname(p))
    except (json.JSONDecodeError, OSError):
        pass
found.sort(key=lambda d: os.path.getmtime(os.path.join(d, 'migration_state.json')))
print(found[-1] if found else '')
" "<OUTPUT_ROOT>"

Phase Playbooks

Every phase's step-by-step procedure lives in its own file under phases/. Read the playbook for the phase you are about to run, run it, then come back here for the next one. Do not preload them all — that is the whole point of the split.

✓ = one of the 9 keys validate_migration_state.py --strict requires for Python. Note the naming trap: Phase 1.2 writes 1a_assessment_report (not 1_2_...). The state key — not the phase number — is the contract with the validator.

Phase Completeness

Because playbooks load one at a time, the phase table above is your only map — work down it in order and do not rely on remembering what comes next.

Before you report the migration complete, run the completeness check (Phase 4a). It is the single authority on whether every phase was navigated:

uv run --project <SKILL_DIRECTORY> \
  python <SKILL_DIRECTORY>/scripts/validate_migration_state.py \
  --state <CONVERSION>/migration_state.json --strict --json

Exit 0 = all 9 required keys present. Exit 1 = at least one phase never ran. Never report success without a green run of this command — it is the canonical success criterion, and skipping it means nothing checked the pipeline end to end.

Two limits of this check you must compensate for by being honest, because it cannot catch you:

• A skip with a reason passes. {"status": "skipped", "skip_reason": "<any text>"} is accepted as satisfied; only a missing key or a skip with no reason fails. So only ever record skipped when the phase genuinely could not run, and put the real cause in skip_reason — never to move past a phase that merely looked hard.
• This validator is itself an optional phase (4a_validation), so nothing forces you to run it. That is exactly why it is listed MUST RUN above.

Per-phase scos_gates.py checks are the complement: they inspect the code on disk, so they catch a phase that ran but did nothing (e.g. unsupported_construct_live, fixed_but_unchanged, phase2_not_orchestrated) — see references/gate-findings.md. State-key presence proves navigation; the gates prove work.

Universal Gate Contract

Every deterministic gate (scripts/scos_gates.py <section> and the phase-specific validators) uses the same exit-code protocol. Read the verdict from stdout — never rely on a non-portable $? capture. A phase playbook only restates this when its retry budget or recovery action differs.

On every FAIL, read references/gate-findings.md before you act. This is a required step, not background reading. The gates emit 35 distinct codes and the code — not the message text — determines what to do. Crucially, several codes are not fixable by a specialist at all: phase2_not_orchestrated and manifest_file_missing mean you skipped a coordinator step, preexisting_syntax is a defect in the customer's source, and sql_mechanical_not_rewritten means Phase 0.6 needs re-running. Re-dispatching a fixer for any of those spends a full pass achieving nothing and risks regressing correct work. The reference gives an Owner per code — fixer / analyzer / reporter / coordinator / source — so check the owner before dispatching anything.

It also records the one known unconvergeable case: convertible_not_converted fires on external-storage I/O whose repoint needs a customer stage name no agent can invent. Never relabel a finding to silence a gate — leave the honest TODO and escalate.

Empty stdout means the gate did not run — it is NOT a FAIL. A malformed command line (missing or invalid --section, unknown flag) exits 2 with empty stdout, because argparse uses exit 2 and does not honour the 3 reserved above. So exit 2 is ambiguous: it means FAIL only when stdout carries a verdict.

Before acting on any exit 2, check that stdout parsed and gaps is non-empty. If stdout is empty, or the verdict is missing, or gaps is empty — fix your own invocation and re-run the gate. Do NOT re-dispatch a specialist. Re-dispatching here spends a full fixer/reporter pass on a failure that does not exist, and with no gaps there is nothing to target, so both retries burn and the gate still never ran. Never record a phase as passed off a gate whose stdout was empty — that phase is unverified, and Phase 4a cannot catch it because it only checks key presence.

Default retry budget is 2 attempts on exit 2, then escalate to the user. Phase 1.2 allows 3. Phase 2b iterates up to 3.

On every advance, record the outcome under migration_state.json :: phases_completed with status, the gate name, and attempts; on an unrunnable phase record {"status": "skipped", "skip_reason": "<one-line reason>"}. You (the coordinator) are the single writer of migration_state.json — specialists return results for you to merge (see phases/phase-2-fixes.md → state-write ownership).

After each phase, git checkpoint inside <CONVERSION>; the playbooks give the exact commit messages and tags (phase-0-source, phase-1-complete, …), which the gates depend on for baselines.

Specialists and References

Resumption

Both cases — continuing from a completed assessment and recovering from mid-migration context loss — are handled by the "Startup — Resume from Existing Assessment" probe above. It scans the output root on every startup, so no explicit handoff parameter is required in either case. Read migration_state.json from the discovered path to restore all coordinator state (manifest, paths, coordinator_mode, pending_files, processed_files).

Stopping Points

• Phase 0: After collecting project info — confirm settings before starting
• Phase 2: If the fixer gate fails after 2 retries — escalate to user with specific errors
• Phase 5: After migration completes — ask user about validation

Success Criteria

• scripts/validate_migration_state.py --strict exits 0 (this is the canonical, machine-checkable success criterion — see Phase 4a)
• migration_state.json shows all phases 1-4 completed with gate approval
• Reports/Issues.csv exists with data rows
• Reports/InputFilesInventory.csv row count matches manifest
• Reports/MigrationReadinessReport.html and Reports/AssessmentIR.json exist
• All .py files pass py_compile syntax check
• Every .py file has a migration header docstring
• File count matches between original and migrated directories

Output

<output_root>/
  Conversion-SCOS-<timestamp>/                       ← <CONVERSION>
    Output/                                          ← <MIGRATED> — converted files
    Reports/
      Issues.csv                                     ← EWI issues (SPRKCNTPY*)
      InputFilesInventory.csv                        ← Source file inventory
      ArtifactDependencyInventory.csv                ← Import dependencies
      MigrationReadinessReport.html                  ← Stakeholder-facing readiness report
      AssessmentIR.json                              ← Structured IR (stable contract)
    Logs/                                            ← Migration log
    migration_state.json                             ← Phase gate tracking
    analysis.json                                    ← Compatibility analysis