Sf crawl orchestrator

sf-crawl-orchestrator — ingestion skill (v1.8 Phase 3, MCP-primary)

9-step protocol. Bridges SF 24 native MCP (HTTP http://127.0.0.1:11435/mcp) to the file-based sf-import skill. MCP-PRIMARY ingestion path (v2.2): the orchestrator handles the full 24-report export per crawl (Tier 1 14 + Tier 2 10) via the build_export_plan() dispatch — each engine canonical name maps to one of three real SF export tools (sf_generate_report, sf_generate_bulk_export, sf_export_seo_element_urls), called with file_path=f"{canonical}.csv" against the currently-loaded crawl — then moves files from SF allowed_directory into projects/{slug}/sf-exports/{date}/raw/, then invokes sf-import as a subprocess for projection into master.xlsx's 6 SF-derived sheets. File-drop fallback is preserved as disaster recovery only.

This skill is the first HTTP MCP consumer in PSEO (D-SF-01 + D-SF-14). Future HTTP MCPs (local LM Studio, custom servers) reuse the scripts.util.sf_mcp_client.SfMcpClient pattern established in Phase 2. SKILL.md body invocations use Claude's mcp__sf__sf_* wrapper form; the companion script scripts/ingestion/sf_crawl_orchestrator.py is pure transform (enumerate_reports + move_with_rollback + parse_progress_response) and does NOT call MCP directly (mirrors gsc_pull.py / dfs_pull.py pattern).

Inputs (frontmatter contract)

Name	Type	Default	Notes
project_slug	string	—	Required. Resolves projects/{slug}/sf-exports/ + project.config.json.
url	string	project.config.domain	Crawl başlangıç URL'si. Omit → fallback to domain.
resume_run_id	string	None	--resume flag; paused workflow_runner run'ı sürdür.
include_tier3	boolean	false	Q-SF-MCP-10 lock: 24 raporu only by default. True → 40 raporu.

workspace_root is resolved via PSEO_WORKSPACE_ROOT env or explicit test override (mirrors workflow_runner / events_writer / sf-import).

Outputs (artifacts produced)

• projects/{slug}/sf-exports/{date}/raw/{report_name}.csv × 24 — exported CSVs after atomic move from SF allowed_directory (D-SF-16 + D-SF-03).
• projects/{slug}/_state/workflows/{run_id}.json — workflow state file (ADR-021); pausable + resumable for mid-loop recovery.
• projects/{slug}/_state/events.jsonl — provenance entries (source.kind=sf_mcp, per-report row).
• projects/{slug}/inbox/sf-mcp/{date}-sf-crawl-{slug}.json — envelope JSON recording crawl_id, report manifest, durations, AMBER warnings (drift recovery witness; mirrors sf-import envelope discipline).
• projects/{slug}/outputs/reports/{date}-sf-crawl.md — human-readable run summary rendered from templates/reports/sf-crawl.template.md.
• Indirect (via sf-import subprocess Step 7): master.xlsx 6 sheet rows.

9-Step Body Protocol

Step name conventions follow sf-import + dfs-pull discipline: workflow_runner.create_run(steps=[...]) carries the 7 workflow-managed step names; create_run itself (Step 1) and complete (Step 9) are body protocol Steps but not entries in the steps[] list.

Step 1 — create_run

Open a workflow run shell. The state file lives at projects/{slug}/_state/workflows/{run_id}.json (ADR-021). If resume_run_id is provided, skip create_run and call workflow_runner.resume instead (paused → running, paused_at preserved per rules/append-only-state.md).

from scripts.state import workflow_runner

if resume_run_id:
    handle = workflow_runner.resume(
        resume_run_id, project_slug=project_slug,
    )
else:
    handle = workflow_runner.create_run(
        skill="sf-crawl-orchestrator",
        project_slug=project_slug,
        steps=[
            {"name": "preflight"},
            {"name": "crawl_trigger"},
            {"name": "poll"},
            {"name": "export_24_reports"},
            {"name": "atomic_move"},
            {"name": "invoke_sf_import"},
            {"name": "emit_provenance_and_report"},
        ],
        initial_status="awaiting_approval",  # Q-SF-MCP-02 lock: requires_approval=true
        approval_meta={
            "approver": "user",
            "subject": f"SF MCP crawl triggered for {project_slug}; 24 raporu export edilecek. Onaylıyor musunuz?",
        },
    )
    # Operator approves via workflow_runner.approve(run_id, approver="user")
    # → awaiting_approval → running; this step resumes from Step 2.

Step 2 — preflight (DURUR-orch-1/2/4/7 enforcement)

Three independent probes:

1. mcp__sf__sf_list_allowed_base_directory returns the SF allowed directory (D-SF-10). If the call raises (GUI not responsive, MCP not connected) → DURUR-orch-1.
2. Compare the returned path to project.config.sf.mcp.allowed_directory; mismatch → DURUR-orch-4.
3. mcp__sf__sf_list_crawls returns the live crawl list. Any entry with status="IN_PROGRESS" → DURUR-orch-7 (R13 concurrent-crawl guard: refuse to trigger a second crawl that would corrupt the GUI state). Note: spec mentions sf_crawl_progress for this check but that tool requires a crawl_id; sf_list_crawls is the natural enumerator.
4. If the SF GUI surfaces an open modal dialog (any prior probe raises IllegalStateException) → DURUR-orch-2.

workflow_runner.start_step(handle.run_id, 0, project_slug=project_slug)
try:
    allowed_dir_resp = mcp__sf__sf_list_allowed_base_directory()
except Exception as exc:
    # orch-1 vs orch-2 distinction: SF GUI modal dialog surfaces as
    # IllegalStateException; everything else (connection refused, timeout)
    # is orch-1. Failure code is the canonical workflow-run.schema enum;
    # DURUR identity travels in the message for operator-facing detail.
    tag = "DURUR-orch-2" if "IllegalStateException" in str(exc) else "DURUR-orch-1"
    workflow_runner.fail(
        handle.run_id, project_slug=project_slug,
        code="mcp_error",
        message=f"{tag} SF MCP allowed_base_directory probe failed: {exc}",
        step_index=0,
    )
    raise SystemExit(2)

mcp_allowed = allowed_dir_resp.get("allowed_directory") if isinstance(allowed_dir_resp, dict) else str(allowed_dir_resp)
expected_allowed = project_config["sf"]["mcp"]["allowed_directory"]
if expected_allowed and mcp_allowed != expected_allowed:
    workflow_runner.fail(
        handle.run_id, project_slug=project_slug,
        code="validation_error",
        message=f"DURUR-orch-4 SF allowed_directory mismatch: MCP={mcp_allowed} expected={expected_allowed}",
        step_index=0,
    )
    raise SystemExit(2)

list_resp = mcp__sf__sf_list_crawls()
in_progress = [c for c in list_resp.get("crawls", []) if c.get("status") == "IN_PROGRESS"]
if in_progress:
    workflow_runner.fail(
        handle.run_id, project_slug=project_slug,
        code="mcp_error",
        message=f"DURUR-orch-7 concurrent SF crawl(s) IN_PROGRESS: {[c.get('crawl_id') for c in in_progress]}",
        step_index=0,
    )
    raise SystemExit(2)  # R13 mitigation

workflow_runner.finish_step(handle.run_id, 0, project_slug=project_slug,
                            output_ref=f"allowed_directory={mcp_allowed}")

Step 3 — crawl_trigger

workflow_runner.start_step(handle.run_id, 1, project_slug=project_slug)
crawl_url = url or project_config["domain"]
crawl_config_path = project_config["sf"]["mcp"].get("crawl_config_path")
trigger_resp = mcp__sf__sf_crawl(
    url=crawl_url,
    **({"crawl_config_file_path": crawl_config_path} if crawl_config_path else {}),
)
crawl_id = trigger_resp["crawl_id"]
workflow_runner.finish_step(handle.run_id, 1, project_slug=project_slug,
                            output_ref=f"crawl_id={crawl_id}")
# Provenance: emit sf_mcp_crawl_started event (informational; full
# provenance batch fires in Step 8 after the run completes).

Step 4 — poll

Loop mcp__sf__sf_crawl_progress(crawl_id) every 60s; bail when status is DONE or FAILED. Max wait project.config.sf.mcp.max_wait_minutes (default 180; Q-SF-MCP-03 lock). Exceeding the cap → DURUR-orch-3 (operator review required).

import time
from scripts.ingestion import sf_crawl_orchestrator

workflow_runner.start_step(handle.run_id, 2, project_slug=project_slug)
max_wait_sec = int(project_config["sf"]["mcp"]["max_wait_minutes"]) * 60
poll_interval = 60
elapsed = 0
final_state = None
while elapsed <= max_wait_sec:
    raw = mcp__sf__sf_crawl_progress(crawl_id=crawl_id)
    state = sf_crawl_orchestrator.parse_progress_response(raw)
    if state.status in ("DONE", "FAILED"):
        final_state = state
        break
    time.sleep(poll_interval)
    elapsed += poll_interval

if final_state is None:
    workflow_runner.fail(
        handle.run_id, project_slug=project_slug,
        code="timeout",
        message=f"DURUR-orch-3 sf_crawl_progress exceeded max_wait_minutes={max_wait_sec // 60}",
        step_index=2,
    )
    raise SystemExit(2)

if final_state.status == "FAILED":
    workflow_runner.fail(
        handle.run_id, project_slug=project_slug,
        code="mcp_error",
        message=f"DURUR-orch-3 sf_crawl_progress reported FAILED for crawl_id={crawl_id}",
        step_index=2,
    )
    raise SystemExit(2)  # terminal-failed variant of orch-3

workflow_runner.finish_step(handle.run_id, 2, project_slug=project_slug,
                            output_ref=f"urls_crawled={final_state.urls_crawled}")

Step 4b — Crawl Analysis (post-crawl, ZORUNLU for dup/hash — v1.7)

near_duplicates + exact_duplicates reports and the internal_all near-dup/hash columns are populated ONLY by SF's post-crawl Crawl Analysis pass — NOT by the crawl itself. If Crawl Analysis has not run, those two reports export with correct headers but 0 data rows (present-but-empty), which is NOT the same as "missing" — do not silently pass it as 24/24-populated. Run Crawl Analysis after poll completes (via MCP if available; otherwise emit an explicit operator instruction: "SF GUI'de Crawl Analysis çalıştır → 2 dup raporunu re-export") and mark the duplicate dimension AMBER until populated. Evidence: a 2026-07-24 run had the Duplicates toggle ON yet both reports empty; after operator ran Crawl Analysis, 2 real 100%-similarity near-dup clusters surfaced (money-page cannibalization).

AUTO-ANALYSIS DETECTION (v1.9 — A20): before flagging AMBER, check sf_crawl_progress → postCrawlAnalysisProgress.percentComplete. If 100, Crawl Analysis ran automatically at crawl-end (a GUI setting) → empty dup/orphan reports are a GENUINE ZERO, not present-but-unanalyzed → report as "genuine 0", do NOT AMBER. Only <100/absent means analysis did not run → then AMBER + operator re-export. Distinguish three states: toggle-off (no report) ≠ analysis-not-run (report present, empty) ≠ genuine-0 (analysis ran, empty). Evidence rkturizm-tr: postCrawlAnalysisProgress=100, both bulk + seo-element paths returned 0 clusters → genuine zero, no re-export needed.

CRAWL CONFIG — USER-AGENT + RENDERING (v1.9 — A19): before the crawl, ensure SF Configuration → User-Agent = Googlebot (Smartphone) (mobile-first indexing → you see what Google actually indexes + catches cloaking/differential-serving; the default "Screaming Frog SEO Spider" UA is served visitor content → misleading audit) and Rendering = Text-Only for server-side stacks (WordPress/RankMath prints schema+H1 server-side → fast + sufficient for element existence; escalate to JS Rendering only if a live-vs-crawl "missing H1/schema" contradiction appears). If an operator asks which UA, recommend Googlebot-Smartphone directly.

Step 5 — export_24_reports (24 raporu × SF-export-tool dispatch loop)

Iterate the export PLAN returned by sf_crawl_orchestrator.build_export_plan(include_tier3=False) — one SfExportSpec per canonical, in the same order as enumerate_reports(include_tier3=False) (14 Tier 1 + 10 Tier 2, sourced from scripts.ingestion.sf_import.TIER1_REQUIRED + TIER2_RECOMMENDED frozensets; SSoT per rules/single-source-of-truth.md). Per-report export goes to a temp staging directory _state/staging/sf-crawl-{run_id}/.

Dispatch contract (Manager live-validated): the engine's 24 canonical report names are NOT Screaming Frog identifiers. Each maps to one of THREE real SF MCP export tools — the SF API keys off category (colon form, e.g. "Links:All Inlinks") or seo_element_name + filter_name (e.g. "Internal" + "All"), NEVER the engine names. There is NO crawl_id, report_name, save_report, or output_directory arg on any of them; the export runs on the currently-loaded crawl (the one Step 3 triggered + Step 4 confirmed DONE) and writes to file_path relative to the SF allowed base directory. We always pass file_path=f"{spec.canonical}.csv" so the file lands with the exact name sf_import.normalize_filename matches downstream.

• sf_generate_report / sf_generate_bulk_export — category + export_type="CSV" (+ file_path).
• sf_export_seo_element_urls — seo_element_name + filter_name (+ file_path). This tool has NO export_type arg → it always emits NDJSON (one flat JSON object per line), even to a .csv path. The export loop converts those 16 reports to CSV in place via sf_crawl_orchestrator.ndjson_to_csv (gated by export_returns_ndjson(spec)) before the atomic move, so sf_import sees a uniform CSV raw/ set.

spec.tool selects which wrapper to call; spec.call_kwargs carries the correct args WITHOUT file_path (we add it). See scripts/ingestion/sf_crawl_orchestrator.SF_EXPORT_DISPATCH for the full 24-entry mapping (one line per canonical, Manager-correctable in place).

Tier policy (matches sf-import):

• Tier 1 fail (spec.tier == "tier1") → DURUR-orch-8: rollback (delete temp staging dir + all partial CSVs), surface RED to operator. D-SF-16 atomic semantics.
• Tier 2 fail (spec.tier == "tier2") → AMBER warning, continue (matches sf-import search_console_all canonical exemption).

import shutil
from pathlib import Path
from scripts.ingestion import sf_crawl_orchestrator

workflow_runner.start_step(handle.run_id, 3, project_slug=project_slug)
export_plan = sf_crawl_orchestrator.build_export_plan(include_tier3=include_tier3)
assert len(export_plan) == (40 if include_tier3 else 24), \
    f"build_export_plan drift: expected {40 if include_tier3 else 24}, got {len(export_plan)}"

temp_staging = workspace_root / "projects" / project_slug / "_state" / "staging" / f"sf-crawl-{handle.run_id}"
temp_staging.mkdir(parents=True, exist_ok=True)

# SF export tools write file_path relative to the allowed base directory; we
# always use "{canonical}.csv" so sf_import.normalize_filename matches.
amber_warnings: list[str] = []
exported: list[str] = []

# Map spec.tool → the live mcp__sf__* wrapper (operator session).
SF_EXPORT_TOOLS = {
    "sf_generate_report": mcp__sf__sf_generate_report,
    "sf_generate_bulk_export": mcp__sf__sf_generate_bulk_export,
    "sf_export_seo_element_urls": mcp__sf__sf_export_seo_element_urls,
}

for spec in export_plan:
    rel_path = f"{spec.canonical}.csv"
    try:
        tool_fn = SF_EXPORT_TOOLS[spec.tool]
        resp = tool_fn(file_path=rel_path, **spec.call_kwargs)
        src = Path(mcp_allowed) / rel_path
        # sf_export_seo_element_urls has NO export_type arg → SF writes NDJSON
        # even to a .csv path (live-verified 2026-06-02). Convert in place so
        # sf_import (CSV-only, header-matched) consumes a uniform CSV raw/ set.
        if sf_crawl_orchestrator.export_returns_ndjson(spec):
            src.write_text(
                sf_crawl_orchestrator.ndjson_to_csv(src.read_text(encoding="utf-8")),
                encoding="utf-8",
            )
        # Move the (now-CSV) export from SF allowed_directory → temp_staging.
        dst = temp_staging / rel_path
        sf_crawl_orchestrator.move_with_rollback(src, dst)
        exported.append(spec.canonical)
    except Exception as exc:
        if spec.tier == "tier1":
            # D-SF-16 rollback: delete temp staging, surface DURUR-orch-8.
            shutil.rmtree(temp_staging, ignore_errors=True)
            workflow_runner.fail(
                handle.run_id, project_slug=project_slug,
                code="mcp_error",
                message=f"DURUR-orch-8 Tier 1 export failed for {spec.canonical!r} "
                        f"(tool={spec.tool} kwargs={spec.call_kwargs}): {exc}; "
                        f"rollback complete (temp_staging deleted)",
                step_index=3,
            )
            raise SystemExit(2)  # D-SF-16 atomic rollback
        # Tier 2 → AMBER, continue.
        amber_warnings.append(
            f"Tier 2 export failed for {spec.canonical!r} "
            f"(tool={spec.tool}): {exc}"
        )

workflow_runner.finish_step(handle.run_id, 3, project_slug=project_slug,
                            output_ref=f"exported={len(exported)} amber={len(amber_warnings)}")

Step 6 — atomic_move (temp staging → projects/{slug}/sf-exports/{date}/raw/)

import datetime
workflow_runner.start_step(handle.run_id, 4, project_slug=project_slug)

today = datetime.date.today().isoformat()
target_raw = workspace_root / "projects" / project_slug / "sf-exports" / today / "raw"
if target_raw.exists():
    workflow_runner.fail(
        handle.run_id, project_slug=project_slug,
        code="validation_error",
        message=f"DURUR-orch-5 sf-exports target already exists: {target_raw}",
        step_index=4,
    )
    raise SystemExit(2)

target_raw.parent.mkdir(parents=True, exist_ok=True)
try:
    shutil.move(str(temp_staging), str(target_raw))  # atomic when same FS
except Exception as exc:
    workflow_runner.fail(
        handle.run_id, project_slug=project_slug,
        code="internal_error",
        message=f"DURUR-orch-6 shutil.move failed: {exc}",
        step_index=4,
    )
    raise SystemExit(2)

workflow_runner.finish_step(handle.run_id, 4, project_slug=project_slug,
                            output_ref=str(target_raw))

Step 7 — invoke_sf_import (Q-SF-MCP-05 default YES — auto-invoke)

Subprocess the existing sf-import skill to project the 24 raw CSVs into master.xlsx's 6 SF-derived sheets. Provenance is chained via sf-import's source_run_id frontmatter input — NOT a CLI flag (passing --source-run-id to the script makes argparse exit 2; see the code note below): sf-import's events.jsonl entry carries this orchestrator's run_id so drift-check can correlate the two events later.

import subprocess
workflow_runner.start_step(handle.run_id, 5, project_slug=project_slug)

result = subprocess.run(
    [
        "python3", "-m", "scripts.ingestion.sf_import",
        "--project", project_slug,
        "--sf-export-path", str(target_raw.parent),  # the {date} dir; sf-import discovers raw/
        # NOTE (live-verified 2026-06-02): the sf_import *script* CLI accepts only
        # --project / --sf-export-path / --workspace-root / --dry-run. source_run_id
        # provenance chaining is an sf-import *skill frontmatter* input (interpreter
        # level), NOT a script flag — passing --source-run-id makes argparse exit 2.
    ],
    capture_output=True, text=True, timeout=600,
)
if result.returncode != 0:
    workflow_runner.fail(
        handle.run_id, project_slug=project_slug,
        code="internal_error",
        message=f"sf-import subprocess returned {result.returncode}: {result.stderr[:500]}",
        step_index=5,
    )
    raise SystemExit(2)

workflow_runner.finish_step(handle.run_id, 5, project_slug=project_slug,
                            output_ref=f"sf_import_exit=0 stdout_tail={result.stdout[-200:]}")

Step 8 — emit_provenance_and_report (sf_mcp source + render template)

One events_writer.append_provenance entry summarizes the entire crawl (per-report row counts captured in outputs of the workflow run; finer- grained per-report rows are NOT emitted to keep events.jsonl compact — the envelope JSON in inbox/sf-mcp/ carries the per-report manifest).

import json
from scripts.state import events_writer
# SSoT tier frozensets (hoisted — used by both the envelope manifest below and
# the report status table further down).
from scripts.ingestion.sf_import import TIER1_REQUIRED, TIER2_RECOMMENDED
workflow_runner.start_step(handle.run_id, 6, project_slug=project_slug)

envelope = {
    "_meta": {
        "captured_at": _utc_iso_z(),
        "tool": "sf_mcp",
        "project_slug": project_slug,
        "crawl_id": crawl_id,
        "crawl_url": crawl_url,
        "exported_count": len(exported),
        "amber_warnings": amber_warnings,
    },
    "reports": [{"canonical_name": r, "tier": "required" if r in TIER1_REQUIRED else "recommended"}
                for r in exported],
}
envelope_path = (
    workspace_root / "projects" / project_slug
    / "inbox" / "sf-mcp"
    / f"{today}-sf-crawl-{project_slug}.json"
)
envelope_path.parent.mkdir(parents=True, exist_ok=True)
envelope_path.write_text(json.dumps(envelope, ensure_ascii=False, indent=2),
                        encoding="utf-8")

events_writer.append_provenance(
    project_id=project_slug,
    source={
        # events.schema source.additionalProperties=false; only these keys are valid:
        "kind": "sf_mcp",
        "mcp_server": "sf",
        # The 24-report export fans out across 3 SF tools (sf_generate_report,
        # sf_generate_bulk_export, sf_export_seo_element_urls) via
        # build_export_plan(); mcp_tool is single-valued, so we record the
        # orchestrator entrypoint. Per-report tool is in the inbox envelope.
        "mcp_tool": "sf__sf_crawl",
        "response_bytes": len(json.dumps(envelope)),
        "row_count": len(exported),
    },
    operation="ingest",
    rows_written=len(exported),
)

# Render report via templates/reports/sf-crawl.template.md.
# render_template.render() uses safe_substitute (see its docstring): any
# template $token not supplied here stays visible in the report (non-fatal)
# rather than discarding a completed 24-report crawl over a cosmetic gap.
from scripts.reporting import render_template

# Status table counters derived from `exported` + the SSoT tier frozensets
# (imported at the top of this step). In
# the success path every Tier 1 report exported (else DURUR-orch-8 rolled the
# run back) so tier1_failed/tier1_missing are 0; Tier 2 misses are the AMBER set.
t1_exported = sum(1 for r in exported if r in TIER1_REQUIRED)
t2_exported = sum(1 for r in exported if r in TIER2_RECOMMENDED)
t2_missing = len(TIER2_RECOMMENDED) - t2_exported

# Per-phase durations are read back from the workflow run's recorded step
# timings (workflow_runner persists started_at/finished_at per step). A phase
# with no recorded duration renders as "n/a" — never a fabricated number.
timings: dict[str, str] = {}  # {"preflight": ..., "poll": ..., "export": ..., "handoff": ..., "total": ...}

report_path = render_template.render(
    template_path=workspace_root / "templates" / "reports" / "sf-crawl.template.md",
    output_path=workspace_root / "projects" / project_slug
                / "outputs" / "reports" / f"{today}-sf-crawl.md",
    variables={
        "project_slug": project_slug,
        "date": today,
        "crawl_id": crawl_id,
        "run_id": handle.run_id,
        "exported_count": str(len(exported)),
        "amber_count": str(len(amber_warnings)),
        "amber_warnings": "\n".join(f"- {w}" for w in amber_warnings) or "_(none)_",
        # 24-report status table (Tier 1 14 required + Tier 2 10 recommended).
        "tier1_exported": str(t1_exported), "tier1_failed": "0", "tier1_missing": "0",
        "tier2_exported": str(t2_exported), "tier2_failed": str(len(amber_warnings)),
        "tier2_missing": str(t2_missing),
        "total_failed": str(len(amber_warnings)), "total_missing": str(t2_missing),
        "tier1_status": "PASS (14/14 required)",
        "tier2_status": f"{t2_exported}/10 exported ({t2_missing} AMBER)",
        # sf-import subprocess handoff result (Step 7 `result`).
        "sf_import_exit_code": str(result.returncode),
        "sf_import_sheet_summary": (result.stdout.strip().splitlines() or ["(no stdout)"])[-1],
        # Durations from the run record (absent phase → "n/a", not fabricated).
        "duration_preflight": timings.get("preflight", "n/a"),
        "duration_poll": timings.get("poll", "n/a"),
        "duration_export": timings.get("export", "n/a"),
        "duration_handoff": timings.get("handoff", "n/a"),
        "total_duration": timings.get("total", "n/a"),
        "recommendations": "_(none)_" if not amber_warnings
                           else "Review the AMBER Tier 2 gaps before the next crawl.",
    },
)

workflow_runner.finish_step(handle.run_id, 6, project_slug=project_slug,
                            output_ref=str(envelope_path))

Step 9 — complete

workflow_runner.complete(handle.run_id, project_slug=project_slug, outputs={
    # F5: outputs.* must be STRING-TYPED.
    "crawl_id": crawl_id,
    "reports_exported": str(len(exported)),
    "amber_warnings": str(len(amber_warnings)),
    "sf_export_path": str(target_raw),
    "envelope_path": str(envelope_path),
    "report_path": str(report_path),
    "sf_import_run_id": handle.run_id,  # same run_id chains both
})

24-Report enumeration (SSoT)

The 24-report list comes from the existing scripts/ingestion/sf_import.py TIER1_REQUIRED (14) + TIER2_RECOMMENDED (10) frozensets, which themselves mirror schemas/sf-required-reports.schema.json definitions.canonicalName.enum. enumerate_reports(include_tier3=False) IMPORTS from these — never re-lists names inline (per rules/single-source-of-truth.md).

When include_tier3=True, the function returns all 40 names from the canonicalName enum minus T1+T2 = 16 Tier 3 entries (future-proofing per Q-SF-MCP-10 lock; default 24).

Tier policy

Tier	Missing/Export fail →	Notes
Required	DURUR-orch-8 + rollback	All 14 must succeed; partial run rolled back.
Recommended	AMBER warning	Matches sf-import policy; search_console_all
		typical exemption.
Optional	SILENT	Only included when include_tier3=True.

D-SF-16 — Atomic crawl semantics

The orchestrator's "all-or-nothing" guarantee on Tier 1 export is implemented by writing every CSV first to a temp directory _state/staging/sf-crawl-{run_id}/. ONLY when all 14 Tier 1 reports export successfully do we shutil.move(temp_staging → sf-exports/{date}/raw/). A single Tier 1 failure triggers shutil.rmtree(temp_staging, ignore_errors=True) BEFORE the orchestrator raises DURUR-orch-8 — sf-import never sees a half-populated raw/ directory.

DURUR conditions (8)

Stop and flag the operator — do not patch, do not fall back.

1. orch-1 — mcp__sf__sf_list_allowed_base_directory raises (SF GUI not responsive, MCP not connected, network error).
2. orch-2 — SF GUI surfaces an IllegalStateException (modal dialog open — e.g. an unsaved settings change). Operator must close the dialog manually before re-running.
3. orch-3 — sf_crawl_progress polling exceeds project.config.sf.mcp.max_wait_minutes (default 180), OR returns status="FAILED" terminally.
4. orch-4 — mcp__sf__sf_list_allowed_base_directory returns a path that mismatches project.config.sf.mcp.allowed_directory (operator must reconcile F-15 isolation governance before proceeding).
5. orch-5 — target projects/{slug}/sf-exports/{date}/raw/ already exists at atomic move time (operator must archive or remove the prior batch; we refuse to overwrite).
6. orch-6 — shutil.move(temp_staging → sf-exports/{date}/raw/) raises (disk full, permission denied, cross-filesystem error). Temp staging is preserved for forensics; operator decides whether to retry or rollback manually.
7. orch-7 — mcp__sf__sf_list_crawls reports an IN_PROGRESS crawl (R13 mitigation; never trigger a parallel crawl that would corrupt the GUI state).
8. orch-8 — Tier 1 export fails for any single report. D-SF-16 atomic rollback: temp_staging deleted, no partial state survives.

Resume capability (D-SF-16 + workflow_runner.pause/resume)

The orchestrator is resumable mid-loop. If mcp__sf__sf_crawl_progress times out OR any of the three SF export tools (sf_generate_report, sf_generate_bulk_export, sf_export_seo_element_urls) raises a recoverable error (transient network), the operator can:

1. Set workflow state to paused: workflow_runner.pause(run_id, ...) (paused_at preserved).
2. After fixing the underlying issue (re-open SF GUI, increase max_wait_minutes), resume via /pseo-sf-crawl <slug> --resume <run_id>.
3. The skill body checks resume_run_id at Step 1 and calls workflow_runner.resume() instead of create_run — preserves step history + previously exported reports in temp staging.

Cross-references

• Schemas: schemas/sf-mcp-tool-mapping.schema.json (Phase 1 NEW; 6 use-case keys + sfMcpTool enum), schemas/sf-required-reports.schema.json (canonical 40-report enum; Tier 1/2/3 frozensets in sf_import.py SSoT), schemas/project-config.schema.json v1.5 (sf.mcp.* block; Migration 0005 populates defaults), schemas/events.schema.json (source.kind=sf_mcp enum addition Phase 1), schemas/skill-frontmatter.schema.json (this frontmatter).
• Cross-modules (IMPORT-only): scripts/state/workflow_runner.py (create_run, start_step, finish_step, complete, fail, pause, resume), scripts/state/events_writer.py (append_provenance, next_run_id), scripts/ingestion/sf_import.py (TIER1_REQUIRED + TIER2_RECOMMENDED frozensets — SSoT for tier membership), scripts/util/sf_mcp_client.py (Phase 2; NOT used here — orchestrator body calls MCP via mcp__sf__sf_* wrapper form; sf_mcp_client is for Phase 5 consumer skills with use_sf_mcp_live=True).
• Tests: tests/skills/test_sf_crawl_orchestrator.py (11 cases: 10 functional — happy_path_24_reports + 8 DURUR cases + sf-import handoff — plus 1 frontmatter-validation bonus), tests/scripts/test_sf_crawl_orchestrator_helpers.py (16 cases: pure-transform helper coverage — enumerate_reports / move_with_rollback / parse_progress_response / build_export_plan / ndjson_to_csv), tests/smoke/test_sf_mcp_smoke.py (1 case live MCP skipif).
• Companion skill: skills/ingestion/sf-import/SKILL.md (frontmatter adds source_run_id optional input in Phase 3; body 8-step protocol UNCHANGED per D-SF-07).
• Command: commands/pseo-sf-crawl.md (Phase 6 NEW; /pseo-sf-crawl <slug> [url] [--resume <run_id>]).

Discipline checklist

• TODO/fallback YASAK — every DURUR raises, none silently downgrade.
• Schema-first — frontmatter validates against schemas/skill-frontmatter.schema.json Draft 7; mcp_tools.required entries match mcp-tool-registry.json sf server inventory (sf_crawl, sf_crawl_progress, sf_generate_report, sf_generate_bulk_export, sf_export_seo_element_urls, sf_list_allowed_base_directory).
• Plugin-agnostik — no slug literals; project_slug flows through.
• ADR-013 — Use when/Also use when/Do not use when are STRING content inside description, not separate fields.
• D-SF-16 — atomic rollback on Tier 1 export failure (temp_staging pattern); sf-import never sees a half-populated raw/ directory.
• D-SF-17 — multi-session execution (this skill = Phase 3 Worker deliverable; Phase 5 consumer flag wiring is a separate Worker).
• F-15 isolation — SF allowed_directory governance preserved (configurable via project.config.sf.mcp.allowed_directory but always cross-checked at preflight).
• F5 — outputs.* values are STRING-TYPED (artifact paths or stringified counts), never raw ints.
• requires_approval=true (Q-SF-MCP-02 lock) — initial_status= "awaiting_approval" in create_run; operator must approve before crawl_trigger fires.
• SSoT — 24-report enumeration imports from scripts.ingestion.sf_import.TIER1_REQUIRED + TIER2_RECOMMENDED frozensets; never re-lists names inline.