Schema Audit

---
name: schema-audit
description: |
  Use when: kullanıcı "schema audit", "structured data", "JSON-LD denetim",
  "schema.org doğrulama", "rich result eksik", "Product schema hatası",
  "BreadcrumbList itemListElement", "FAQPage validation" der ya da
  /pseo-schema-audit çağırır.
  Also use when: aktif projenin SF (Screaming Frog) export'u alınmış
  (`projects/{slug}/sf-exports/{date}/raw/structured_data_all.csv`),
  sf-import skill çalışmış ve `inbox/sf/{date}-{slug}.json` envelope'i
  hazır; on-page-audit / tech-audit ile birlikte triage; rich result
  eligibility kontrol edilecek; opsiyonel olarak DFS content_parsing ile
  canlı schema cross-validate yapılacak.
  Do not use when: SF export henüz yok (sf-import önce çalışmalı, DURUR
  #1); GSC kannibalizasyon (cannibalization), pozisyon 11-20 fırsat
  taraması (quick-wins), title/meta coverage (on-page-audit), tech-seo
  issues (tech-audit) — ayrı discovery skill'leri. Master.xlsx yokken
  çağırma; init-project önce çalışmalı (DURUR #6).
version: "1.1"
status: active
category: discovery
inputs:
  project_slug:
    type: string
    required: true
    description: "Slug; resolves projects/{slug}/master.xlsx + project.config.json + sf-exports/."
  sf_export_date:
    type: string
    required: false
    description: "SF export date (YYYY-MM-DD); defaults to most recent dir under projects/{slug}/sf-exports/."
  default_status:
    type: string
    required: false
    default: "TODO"
    description: "statusEnum seed value when SF detects type-only (no JSON-LD blob); overridden by per-row gap analysis."
  cross_validate_dfs:
    type: boolean
    required: false
    default: false
    description: "Optional DFS on_page_content_parsing live cross-validate (paid; ~3 credits/URL)."
  strict_parse:
    type: boolean
    required: false
    default: false
    description: "When true, malformed JSON-LD raises JsonLdParseError (default: drop row, keep going)."
  use_sf_mcp_live:
    type: boolean
    required: false
    default: false
    description: "Opt-in (D-SF-11): when true, calls SF MCP via scripts/util/sf_mcp_client.SfMcpClient for live structured_data_all rows (bypasses file requirement for fresh JSON-LD). Resolves the crawl id from sf_list_crawls (domain match → instanceDirName), client.load_crawl(...) (resilient), then exports structured_data_all via SF_EXPORT_DISPATCH (sf_export_seo_element_urls 'Structured Data') to file_path (the >100KB inline cap is resolved by writing to disk, not a non-existent 'truncated' flag); the seo-element export is NDJSON → converted via sf_crawl_orchestrator.ndjson_to_csv → csv.DictReader → transform(raw_sf={'rows': ...}). Requires SF GUI + MCP server running (preflight via client.health(); on FAIL / no matching crawl / SfMcpToolError / load timeout → AMBER fallback to file-based path, NEVER hard fail per R9)."
  merchant_checks:
    type: boolean
    required: false
    description: "Merchant listing checks M1–M7 (Step 4c, pure compute, 0 credits). NO static default: omitted → auto-resolves from project.config.profiles (on when the project declares the e-commerce profile, off otherwise). Explicit true/false overrides the profile gate. Rows ride master.xlsx#schema; rules/merchant-structured-data.md governs framing."
  price_parity_sample:
    type: boolean
    required: false
    default: false
    description: "M7 flag: sample ≤10 product URLs (free Scrapling bulk_get, pre-fetched by the orchestrator) and compare JSON-LD price vs rendered page text (TR number formats tolerated). Heuristic — default off; only meaningful when merchant checks run."
outputs:
  - "master.xlsx#schema"
  - "outputs/reports/{date}-schema-audit.md"
  - "events.jsonl"
  - "inbox/sf/{date}-schema-{slug}.json"
consumes:
  - "init-project:projects/{slug}/master.xlsx"
  - "sf-import:projects/{slug}/inbox/sf/{date}-{slug}.json"
produces:
  - "drift-check"
  - "monthly-report"
triggers:
  manual: ["/pseo-schema-audit"]
  natural_language: |
    "schema audit", "structured data audit", "JSON-LD denetim",
    "schema.org doğrulama", "rich result eksik", "Product schema",
    "BreadcrumbList itemListElement", "FAQPage validation",
    "Article headline missing"
  hooks: []
  scheduled:
    - cron: "0 9 * * 3"
      mode: "report-only"
mcp_tools:
  required: []
  optional:
    - "mcp__dataforseo__on_page_content_parsing"
    - "mcp__sf__sf_list_crawls"
    - "mcp__sf__sf_load_crawl"
    - "mcp__sf__sf_export_seo_element_urls"
    - "mcp__ScraplingServer__fetch"
    - "mcp__ScraplingServer__bulk_get"
budget:
  uses_paid_mcp: false
  estimated_credits: 0
autonomy:
  confidence: HIGH
  requires_approval: true
  safe_auto_execute: false
---

# schema-audit — discovery skill (Phase 7 Wave 2)

10-step protocol. Steps map 1:1 to `workflow_runner` invocations + the
spec §16.5 8-step MCP discipline. Raw SF data drift recovery is
mandatory: the parsed SF structured-data export is dropped into
`inbox/sf/` *before* any transform runs, so a transform bug never costs
us the upstream payload.

This skill follows the **convention authority** established by
`skills/discovery/cannibalization/SKILL.md` (Phase 7 Wave 1) +
`skills/ingestion/sf-import/SKILL.md` (Phase 5 Wave 2) +
`skills/discovery/quick-wins/SKILL.md` (Phase 5). The 10-step protocol
shape, raw inbox discipline, URL normalization (D-03), DURUR + flag
rule, and provenance event format are reused verbatim — only the domain
content (target sheet, transform script, schema-validity heuristic)
changes. Deviate only with an ADR.

## Inputs (frontmatter contract)

| Name                  | Type    | Default | Notes                                                              |
|-----------------------|---------|---------|--------------------------------------------------------------------|
| `project_slug`        | string  | —       | Required. Resolves `projects/{slug}/master.xlsx`.                   |
| `sf_export_date`      | string  | latest  | Pins which SF export under `projects/{slug}/sf-exports/{date}/`.    |
| `default_status`      | string  | "TODO"  | statusEnum seed; per-row gap analysis can override per row.         |
| `cross_validate_dfs`  | boolean | false   | When true, runs DFS on_page_content_parsing cross-validate (paid).  |
| `strict_parse`        | boolean | false   | Strict JSON-LD parse mode (raise on malformed blob).                |
| `merchant_checks`     | boolean | auto    | Merchant M1–M7 (Step 4c). Omitted → on iff `e-commerce` ∈ profiles. |
| `price_parity_sample` | boolean | false   | M7: ≤10 URL price-parity sample over pre-fetched HTML (heuristic).  |

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

## Outputs (artifacts produced)

- `projects/{slug}/master.xlsx#schema` — one row per (schema_type,
  scope) tuple (5 cols, schema-locked).
- `projects/{slug}/outputs/reports/{date}-schema-audit.md` —
  human-readable summary (top types, BLOCKED conflicts, missing-required
  hotspots).
- `projects/{slug}/_state/events.jsonl` — `event_kind=provenance`
  entries (`source.kind=sf_csv` always, plus `dataforseo_mcp` when the
  optional cross-validate branch fires; `target_excel_sheet=schema`).
- `projects/{slug}/inbox/sf/{date}-schema-{slug}.json` — parsed SF
  structured-data envelope (drift recovery; flat `{"rows":[...]}` shape).
- (optional) `projects/{slug}/inbox/dfs/{date}-content_parsing-schema-{slug}.json`
  — DFS payload when cross-validate ran.

## 10-Step Body Protocol

> Each step name must match the `steps[*].name` passed to
> `workflow_runner.create_run`. Names are stable identifiers across runs.

### Step 1 — `create_run`

Open a workflow run shell. The state file lives at
`projects/{slug}/_state/workflows/{run_id}.json` (ADR-021, ADR-019).

```python
from scripts.state import workflow_runner
handle = workflow_runner.create_run(
    skill="schema-audit",
    project_slug=project_slug,
    steps=[
        {"name": "validate_sf_inputs"},
        {"name": "parse_sf_structured_data"},
        {"name": "transform"},
        {"name": "optional_dfs_cross_check"},
        {"name": "request_approval"},
        {"name": "write_excel"},
        {"name": "render_report"},
    ],
)
```

### Step 2 — `validate_sf_inputs` (DURUR #1 gate)

Resolve the SF export dir. If `sf_export_date` is omitted, pick the
most recent date subfolder under `projects/{slug}/sf-exports/`.
Required source: `structured_data_all.csv` (or its normalized variant
from sf-import). Missing → DURUR #1.

```python
sf_root = workspace_root / "projects" / project_slug / "sf-exports"
if not sf_root.is_dir():
    workflow_runner.fail(
        handle.run_id, project_slug=project_slug,
        code="validation_error",
        message="sf-exports/ missing — run sf-import first",
        step_index=0,
    )
    raise SystemExit(2)  # DURUR #1
```

### Step 3 — `parse_sf_structured_data` (raw → drift-recoverable JSON)

Load the SF structured-data CSV via `schema_audit_transform.load_sf_csv`,
or pull the `structured_data_all` rows out of the sf-import envelope at
`inbox/sf/{date}-{slug}.json`. Drop the parsed flat envelope into
`inbox/sf/{date}-schema-{slug}.json` BEFORE the transform runs.

```python
from scripts.discovery import schema_audit_transform as sat
sf_csv = sf_root / sf_export_date / "raw" / "structured_data_all.csv"
flat_envelope = sat.load_sf_csv(sf_csv)
inbox_path = (
    workspace_root / "projects" / project_slug
    / "inbox" / "sf"
    / f"{today.isoformat()}-schema-{project_slug}.json"
)
inbox_path.parent.mkdir(parents=True, exist_ok=True)
inbox_path.write_text(
    json.dumps(flat_envelope, ensure_ascii=False, indent=2),
    encoding="utf-8",
)
```

### Step 4 — `transform`

Pure compute via `scripts/discovery/schema_audit_transform.py`:

```bash
python3 scripts/discovery/schema_audit_transform.py \
    --raw-sf inbox/sf/{date}-schema-{slug}.json \
    [--strict-parse] \
    [--default-status TODO] \
    --output-dir _state/transform/{run_id}/
```

`--default-status` (default `TODO`) seeds the statusEnum value for
type-only rows (SF detected the schema type but no JSON-LD blob, so props
can't be verified); it threads to `transform(..., default_status=...)`.

Produces a JSON array (`schema`) shaped to the master-excel schema
(5 columns: schema_type, status, location, scope, remaining_work). URL
normalization (D-03) is applied here, not at parse time, so the inbox
copy is byte-faithful to the SF source. The transform is idempotent:
same input → byte-identical output.

If the parsed instance count is 0 (no JSON-LD blob, no Type-only
fallback), the skill SKIPS write_excel and emits a "no schema markup
detected" notice (DURUR #9 — clean exit, not error).

### Step 4b — `optional_dfs_cross_check` (only when `cross_validate_dfs=true`)

When the optional branch is invoked, **first** run the budget
pre-flight via `scripts/budget/check_budget.py` against the project's
24h DFS credit total.

- Budget pre-flight PASS → fetch DFS `on_page_content_parsing` per URL,
  persist raw to `inbox/dfs/{date}-content_parsing-schema-{slug}.json`,
  re-run `transform(raw_sf, raw_dfs)` so `remaining_work` is decorated
  with "(DFS confirmed live)" / "(DFS did not echo live — verify
  rendering)" notes.
- Budget pre-flight FAIL → **DURUR #4** (`BudgetExceededError`). Do NOT
  silently fall back to SF-only. The skill exits with
  `awaiting_approval`; the manager must explicitly authorize either
  raising the budget cap or skipping the cross-check (recorded as an
  ADR), then resume.

```python
from scripts.discovery import schema_audit_transform as sat
if cross_validate_dfs:
    estimate = sat.estimate_credits(url_count=len(distinct_urls))
    envelope = sat.preflight_budget(
        estimated_credits=estimate,
        project_config_path=str(project_root / "project.config.json"),
        events_path=str(project_root / "_state" / "events.jsonl"),
    )
    # PASS → fetch DFS, persist raw, re-run transform with raw_dfs.
```

The default frontmatter declaration is `uses_paid_mcp: false`,
`estimated_credits: 0`. The optional branch is the **only** path that
incurs DFS credits, and it is gated by an explicit input + budget
pre-flight (Q-W-A4-01 lesson — no per-call/per-url leakage in
frontmatter).

### Step 4c — `merchant_listing_audit` (pure compute, profile-gated)

Merchant listing experience checks M1–M7 per
`rules/merchant-structured-data.md` (offer accuracy + shipping/returns
org-level-first). Pure compute via
`scripts/discovery/merchant_listing_audit.py` over the SAME parsed SF
envelope — 0 credits, no new MCP requirement. Gate: explicit
`merchant_checks` wins; omitted → auto-on iff `"e-commerce"` is in
`project.config.profiles`.

> Placement note: the gap spec drafted this as "Step 6b" (post-approval).
> It runs as Step 4c — BEFORE Step 5 — so the approval subject's row
> count covers the merchant rows; injecting rows after the operator
> approved a smaller count would hollow out the consent gate.

```python
from scripts.discovery import merchant_listing_audit as mla

run_merchant = (
    merchant_checks
    if merchant_checks is not None
    else "e-commerce" in project_config.get("profiles", [])
)
merchant_md = (
    "Merchant listing kontrolleri: atlandı (merchant_checks kapalı / "
    "profil e-commerce değil)."
)
if run_merchant:
    jsonld_index = mla.build_jsonld_index(flat_envelope)
    org_nodes = mla.collect_org_nodes(jsonld_index)
    # Org markup surface (homepage/policy pages) is normally already in
    # the SF export. If org_nodes is empty, OPTIONALLY do ONE free
    # mcp__ScraplingServer__fetch of the homepage ("/"), parse its JSON-LD
    # <script> blocks into nodes and extend org_nodes; on ANY fetch
    # failure just continue — M4/M5 then carry the explicit
    # "not observable" wording instead of asserting absence.
    parity_html_by_url = None
    if price_parity_sample:
        # M7: ≤10 product URLs via free mcp__ScraplingServer__bulk_get;
        # pass the {url: html} dict through — the module does NO I/O and
        # caps the sample at mla.PARITY_SAMPLE_CAP itself.
        parity_html_by_url = fetched_html_by_url
    merchant_result = mla.audit_merchant_listings(
        jsonld_index, org_nodes, project_config,
        today=today.isoformat(),
        rendered_html_by_url=parity_html_by_url,
        price_parity_sample=price_parity_sample,
    )
    schema_rows = schema_rows + merchant_result["rows"]  # no mutation
    merchant_md = mla.render_merchant_findings_md(merchant_result)
```

Merchant rows are TODO-seeded 5-col `schema` sheet rows (priority words
encoded in `remaining_work`, e.g. `merchant M1/high: …` — the sheet has
no severity column by design). They are written by the SAME Step-7
`committer.commit` call and covered by the SAME Step-9 `sf_csv`
provenance event (`rows_written` includes them; the module itself writes
no events — D-003). Shipping/returns gaps yield ONE site-level
opportunity row each, never per-product spam; recommendation defaults to
org-level `Organization.hasShippingService` → `ShippingService` +
`Organization.hasMerchantReturnPolicy` (single policy-page script — the
only deploy mechanism limited-template TR platforms reliably support),
with per-offer `OfferShippingDetails`/`hasMerchantReturnPolicy` reserved
for genuine per-product deviations.

### Step 5 — `request_approval` (skill EXIT awaiting_approval)

```python
workflow_runner.request_approval(
    handle.run_id, project_slug=project_slug,
    approver="user",
    subject=f"{len(rows)} schema audit satırı bulundu, master.xlsx#schema'ya yazalım mı?",
    step_index=4,
)
# Skill exits here. The user replies in a fresh session; resume below.
```

### Step 6 — Resume (`approve` → continue)

```python
workflow_runner.approve(handle.run_id, project_slug=project_slug,
                        approver="user")
```

### Step 7 — `write_excel` (atomic, schema-validated)

Single `committer.commit` call for the schema sheet — the orchestrator's
idempotent commit path (whole-block `transaction.replace` from the schema's
`data_start_row`, so re-running never duplicates rows on the `schema`
snapshot sheet). Goes through the single approved write path with backup,
lock, schema validation, and post-write provenance event emission. **Note:**
the transform module itself does NOT import `scripts.excel.transaction` —
only the orchestrator layer does (via `committer`; cross-module IMPORT
discipline).

```python
from scripts.orchestration import committer
committer.commit(
    workspace_root/"projects"/project_slug/"master.xlsx",
    "schema",
    schema_rows,
    run_id=handle.run_id,
    project_slug=project_slug,
    writer="schema-audit",
)
```

### Step 8 — `render_report`

`render_template.py templates/reports/schema-audit.template.md
data.json` → `outputs/reports/{date}-schema-audit.md`. Variables:
`$project_slug`, `$date`, `$row_count`, `$blocked_count`, `$todo_count`,
`$exists_count`, `$done_count`, `$top_schema_type`,
`$top_remaining_work`, `$report_summary`, `$merchant_findings_md`.

`$merchant_findings_md` is ALWAYS provided (the CLI renderer substitutes
strictly — a missing key aborts): when Step 4c ran it is
`mla.render_merchant_findings_md(merchant_result)`, otherwise the
skipped-state string from the Step 4c snippet. The block never contains
bare rule-id tokens (template R-token lock) — it cites
`rules/merchant-structured-data.md` by path.

### Step 9 — Provenance event

One provenance entry for the SF parse (always), plus one for DFS when
the optional branch fires. Both target `target_excel_sheet=schema`.

```python
from scripts.state import events_writer
events_writer.append_provenance(
    project_id=project_slug,
    source={
        "kind": "sf_csv",
        "source_folder": str(sf_root.relative_to(workspace_root)),
        "row_count": meta["sf_row_count"],
    },
    operation="project_excel",
    target_excel_sheet="schema",
    rows_written=len(schema_rows),
)
# Optional: when DFS branch ran, second event with kind="dataforseo_mcp"
# + cost.credits = estimate, target_excel_sheet=schema (same sheet).
```

### Step 10 — `complete`

```python
workflow_runner.complete(handle.run_id, project_slug=project_slug, outputs={
    # F5: outputs.* must be STRING-TYPED.
    "row_count": str(len(schema_rows)),
    "blocked_count": str(sum(1 for r in schema_rows if r["status"] == "BLOCKED")),
    "todo_count":    str(sum(1 for r in schema_rows if r["status"] == "TODO")),
    "report_path":   str(report_path),
    "raw_inbox":     str(inbox_path),
})
```

## SF MCP Live Mode (Optional, `use_sf_mcp_live=true` — D-SF-11)

**Default: `use_sf_mcp_live=false`** — file-based ingestion (SF CSV via
sf-import) is the canonical contract. The flag is opt-in only and gated
by a `client.health()` preflight; on probe failure the run AMBER-warns
and continues with the file-based path (never hard-fail per R9
mitigation).

When `use_sf_mcp_live=true`, the SKILL body branches at Step 3
(`parse_sf_structured_data`) to optionally bypass the on-disk SF CSV
requirement and pull `structured_data_all` rows inline from SF MCP.

**This branch was rewritten (AC-13 replication) against the REAL SF MCP
API** — the same class of fix as tech-audit (landed in `6ee6fb9`,
live-proven on the demo-aluminum crawl) and the sf-crawl-orchestrator
Step-5 export dispatch (`a714e43`). The engine canonical
`structured_data_all` is NOT a Screaming Frog identifier: it maps via
`sf_crawl_orchestrator.SF_EXPORT_DISPATCH` to the real
`sf_export_seo_element_urls` tool with `seo_element_name="Structured
Data"` + `filter_name="All"`. There is NO `crawl_id`, `report_name`, or
`save_report` arg; the export runs on the *currently-loaded* crawl and is
written to `file_path` (relative to the SF allowed base directory). The
real SF REJECTS inline output >100KB with a tool error — so `file_path`
(write-to-disk) is the canonical export path (OQ-FILEPATH-EXPORTS), not a
non-existent `truncated` flag.

The `sf_list_crawls` result comes back through `SfMcpClient.call_tool`
as the MCP content envelope `{"isError":false,"content":[{"text":
"<JSON>"}]}`, so the crawl list is JSON-encoded inside `content[*].text`
(exactly like `sf_crawl_progress`). We decode the first JSON block, then
match on `url` → `instanceDirName`. The real entry shape (Manager-probed)
is `{"url":"https://demo-aluminum.example/","instanceDirName":"fc718e3f-..."}`.

**Critical: a seo-element export is NDJSON, not CSV.**
`sf_export_seo_element_urls` has no `export_type` arg and always writes
one flat JSON object per line (live-verified — `export_returns_ndjson`
semantics). So the written file is converted to CSV text first via
`sf_crawl_orchestrator.ndjson_to_csv(...)` (which emits a clean CSV
header — no BOM), then parsed with `csv.DictReader`.

```python
import csv
import json
from pathlib import Path
from scripts.ingestion import sf_crawl_orchestrator

def _decode_sf_envelope(result: dict):
    """Decode the first JSON payload from an SfMcpClient content envelope.

    SfMcpClient.call_tool returns {"isError":..,"content":[{"text":"<JSON>"}]}.
    Returns the parsed object (list/dict) or None if no JSON block is present.
    """
    for block in (result or {}).get("content", []) or []:
        if isinstance(block, dict) and isinstance(block.get("text"), str):
            try:
                return json.loads(block["text"])
            except ValueError:
                continue
    return None

def _norm_domain(u: str) -> str:
    """Normalise a URL for domain matching: drop scheme + leading www. +
    trailing slash."""
    bare = (u or "").strip().rstrip("/").split("://", 1)[-1].lower()
    return bare[4:] if bare.startswith("www.") else bare

amber_warnings: list[str] = []
if use_sf_mcp_live:
    from scripts.util.sf_mcp_client import (
        SfMcpClient, SfMcpToolError, SfMcpError,
    )
    sf_cfg = project_config["sf"]["mcp"]
    client = SfMcpClient(base_url=sf_cfg["url"])
    allowed_dir = Path(sf_cfg["allowed_directory"])  # SF write base (D-SF-10)
    if not client.health():
        # R9 AMBER fallback — SF MCP unreachable, continue file-based.
        amber_warnings.append(
            "SF MCP unavailable (health probe failed); "
            "falling back to file-based path"
        )
    else:
        try:
            # 1) Resolve the crawl id: pick the crawl whose `url` matches the
            #    project domain (normalise scheme + leading www. + trailing
            #    slash), then take its `instanceDirName` (the real crawl id).
            list_resp = client.call_tool("sf_list_crawls")
            decoded = _decode_sf_envelope(list_resp)
            crawls = decoded if isinstance(decoded, list) else (
                decoded.get("crawls", []) if isinstance(decoded, dict) else []
            )
            target = _norm_domain(project_config["domain"])
            match = next(
                (c for c in crawls if _norm_domain(c.get("url", "")) == target),
                None,
            )
            if match is None:
                # R9 AMBER fallback — no crawl for this domain; do NOT hard fail.
                amber_warnings.append(
                    f"No SF crawl found for domain {project_config['domain']!r} "
                    "in sf_list_crawls; falling back to file-based path"
                )
            else:
                crawl_id = match["instanceDirName"]
                # 2) Ensure it is the active loaded crawl (resilient: tolerates
                #    the client-side load timeout, polls progress to confirm).
                client.load_crawl(crawl_id)

                # 3) Export "Structured Data" via the orchestrator dispatch
                #    (structured_data_all → ("sf_export_seo_element_urls",
                #    {"seo_element_name":"Structured Data","filter_name":
                #    "All"})). The >100KB inline cap is resolved by file_path.
                tool, call_kwargs = sf_crawl_orchestrator.SF_EXPORT_DISPATCH[
                    "structured_data_all"
                ]
                rel_path = "structured_data_all.ndjson"
                client.call_tool(tool, file_path=rel_path, **call_kwargs)

                # 4) A seo-element export is NDJSON (no export_type arg) → convert
                #    to clean CSV text first (ndjson_to_csv emits a BOM-free
                #    header), then parse with csv.DictReader. The transform
                #    consumes the same flat `{"rows": [...]}` envelope shape.
                csv_text = sf_crawl_orchestrator.ndjson_to_csv(
                    (allowed_dir / rel_path).read_text(encoding="utf-8")
                )
                rows = list(csv.DictReader(csv_text.splitlines()))
                flat_envelope = {"rows": rows}
        except (SfMcpToolError, SfMcpError) as exc:
            # R9 AMBER fallback — list/load/export failed (4xx/5xx, JSON-RPC
            # error, load_crawl settle timeout). NEVER hard fail.
            amber_warnings.append(
                f"SF MCP error: {exc}; falling back to file-based path"
            )

# Pass the parsed rows straight into the transform (the same flat
# envelope the file-based path builds from sf-import).
result = schema_audit_transform.transform(raw_sf=flat_envelope, raw_dfs=raw_dfs)
```

**AMBER vs RED policy (R9 contract — never hard-fail this branch):**

- `client.health()` returns False → AMBER warning, continue file-based.
- No crawl matches the project domain in `sf_list_crawls` → AMBER
  warning, continue file-based.
- `SfMcpToolError` / `SfMcpError` raised (export 4xx/5xx, JSON-RPC error,
  `load_crawl` settle timeout) → AMBER warning, continue file-based.
- A partial SF outage degrades to the file-based
  `structured_data_all.csv` (the transform tolerates partial rows).
- NEVER raise `SystemExit` from this branch. RED reserved for the
  existing DURUR set (SF data missing in BOTH paths, schema drift, etc.).

**Tool naming reminder:** `call_tool(tool_name=...)` takes the **native**
SF MCP tool name (`"sf_export_seo_element_urls"`, `"sf_list_crawls"`,
`"sf_load_crawl"`), NOT the registry form
(`"sf__sf_export_seo_element_urls"`) or the Claude Code wrapper form
(`"mcp__sf__sf_export_seo_element_urls"`). JSON-RPC `params.name` follows
the native form per MCP spec. `SfMcpClient` already speaks the MCP
Streamable-HTTP transport (session handshake + SSE), so the body only
ever passes native tool names + the SF call kwargs.

`amber_warnings` is surfaced via the existing provenance event
`source.kind=sf_mcp` (Step 9) and the rendered report template
variable `$amber_warnings`.

## Schema-validity heuristic (transform domain)

For each parsed JSON-LD / microdata instance:

```
1. Coerce @type → schema_type (lists → first non-empty entry).
2. If raw @type was a list with ≥2 distinct entries → status=BLOCKED
   (schema.org allows multi-type, but mixed top-level types break rich
    results in practice).
3. Otherwise look up _REQUIRED_PROPS[schema_type]:
     - Required props all present → check _RECOMMENDED_PROPS:
         - All recommended present → status=DONE.
         - Some missing            → status=EXISTS, remaining_work names them.
     - Required missing            → status=TODO, remaining_work names them.
   Type-only fallback (presence detected, no blob) → status=EXISTS with
   "verify required props" remaining_work.
4. DFS cross-validate (when present) only DECORATES remaining_work
   ("(DFS confirmed live)" / "(DFS did not echo live)") — it never
   silently flips a status.
5. Aggregate: when ≥3 URLs share the same (schema_type, status,
   remaining_work) signature, collapse into a single site-wide row
   whose location is "{N} URLs" and scope is "site-wide".
```

### Required-prop minimums

```
Article / NewsArticle / BlogPosting   headline, author, datePublished
Product                               name, image, offers
Organization                          name, url
LocalBusiness                         name, address, telephone
BreadcrumbList                        itemListElement
FAQPage                               mainEntity
Recipe                                name, recipeIngredient, recipeInstructions
Event                                 name, startDate, location
Person                                name
WebSite / WebPage                     name (+ url for WebSite)
```

Extend via ADR rather than silently drift.

## URL normalization (D-03 invariant)

Every URL in this skill flows through `_normalize_url` (mirrors
`cannibalization_transform.normalize_url` semantics) so cross-sheet
joins (cannibalization ↔ schema, on_page_audit ↔ schema) remain
bit-stable. Lowercase scheme+host, IDN→punycode, strip default ports,
strip trailing slash (root excluded), drop fragment, drop tracking
params, sort remaining query keys.

## DURUR conditions (8)

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

1. **SF data missing** — `projects/{slug}/sf-exports/` empty OR
   `structured_data_all.csv` absent OR sf-import envelope unreadable.
   STOP, surface "run sf-import first".
2. `master.xlsx#schema` column count or names don't match schema
   (`schemas/master-excel.schema.json#schema`, 5 cols). STOP, schema-first
   violation.
3. JSON-LD parse error in `strict_parse=true` mode → `JsonLdParseError`
   bubbles up, transform aborts. (Default `strict_parse=false` mode
   silently drops malformed rows; the inbox copy is the durable witness.)
4. Optional DFS cross-validate invoked AND `preflight_budget()` FAILs
   → `BudgetExceededError`. STOP, awaiting_approval; do NOT fall back to
   SF-only.
5. `workflow_runner.create_run` fails schema validation
   (`schemas/workflow-run.schema.json`). STOP.
6. `PSEO_WORKSPACE_ROOT` env var unset and no explicit `workspace_root`
   arg passed to `workflow_runner` / `events_writer`. STOP, surface to
   manager.
7. `transaction.append` raises `RowSchemaError` (e.g., `status` not in
   statusEnum, column tuple drift). STOP.
8. statusEnum lookup drift — emitted status not in the canonical 7
   values (`TODO`/`ONGOING`/`EXISTS`/`DONE`/`BLOCKED`/`DEFERRED`/`CANCELED`).
   STOP. Self-checked at row-emission time inside `_validate_row_shape`.

(Adjacent clean-exit case, NOT a DURUR: when transform yields 0 schema
rows the skill goes straight to `complete` with `row_count="0"` and
skips write_excel + render_report — same shape as cannibalization
DURUR #7 clean exit.)

## Cross-references

- Schemas: `schemas/master-excel.schema.json` (schema sheet, 5
  required_columns + `#/definitions/statusEnum`),
  `schemas/events.schema.json` (`source.kind=sf_csv|dataforseo_mcp`,
  `target_excel_sheet=schema`),
  `schemas/sf-required-reports.schema.json` (canonical_name
  `structured_data_all`),
  `schemas/skill-frontmatter.schema.json` (this frontmatter).
- Cross-modules (IMPORT-only): `scripts/state/workflow_runner.py`,
  `scripts/excel/transaction.py`, `scripts/state/events_writer.py`,
  `scripts/budget/check_budget.py`,
  `scripts/reporting/render_template.py`.
- Transform: `scripts/discovery/schema_audit_transform.py` (pure;
  zero `transaction.append` direct calls, zero
  `scripts.excel.transaction` imports).
- Merchant module (Step 4c): `scripts/discovery/merchant_listing_audit.py`
  (pure, deterministic — `today` injected; same import discipline) +
  `rules/merchant-structured-data.md`.
- Tests: `tests/skills/test_schema_audit.py` (≥6 cases incl. budget +
  smoke E2E + statusEnum coverage),
  `tests/discovery/test_merchant_listing_audit.py` (M1–M7 contract),
  `tests/rules/test_merchant_structured_data_rules.py` (rule sentinels +
  template R-token lock).
- Template: `templates/reports/schema-audit.template.md`
  (incl. `$merchant_findings_md`).

## Discipline checklist

- [x] TODO/fallback YASAK — every DURUR raises, none silently downgrade
      (exception: 0-row case is a clean no-op exit, mirrors
      cannibalization DURUR #7).
- [x] Schema-first — frontmatter validates against
      `schemas/skill-frontmatter.schema.json` Draft 7. Budget block:
      `uses_paid_mcp: false`, `estimated_credits: 0` only;
      `_per_call`/`_per_url` keys ABSENT (Q-W-A4-01 lesson).
- [x] Plugin-agnostik — no slug literals; `project_slug` flows through
      every path; transform has 0 hardcoded slug words.
- [x] ADR-013: `Use when`/`Also use when`/`Do not use when` are STRING
      content inside `description`, not separate fields.
- [x] Cross-module IMPORT discipline — `transaction` /
      `workflow_runner` / `events_writer` / `check_budget` are imported,
      never modified from this skill. Transform has 0 imports from
      `scripts.excel.transaction`.
- [x] F1: write target is `master.xlsx` (lowercase, schema-shaped).
- [x] F5: `outputs.*` values are STRING-TYPED (artifact paths or
      stringified counts), never raw ints.
- [x] Append-only state — `events.jsonl` only grows; no in-place rewrite.
- [x] statusEnum strict — every emitted row's status is in the canonical
      7-value enum, validated at row-emission time.

## Changelog

- **v1.1 (2026-06-10, GAP-A3 / GAP-A-B1):** merchant listing module
  (M1–M7) added as Step 4c — pure compute via
  `scripts/discovery/merchant_listing_audit.py`, profile-gated
  (`merchant_checks` auto-on for e-commerce), 0 credits. Report gains
  `$merchant_findings_md`; `rules/merchant-structured-data.md` created.
  Deviation from the gap spec recorded: spec drafted the step as
  "Step 6b" (after approval); placed at 4c so the Step-5 approval count
  covers merchant rows (consent integrity).
- **v1.0:** initial 10-step protocol (Phase 7 Wave 2).