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\",\n\"schema.org doğrulama\", \"rich result eksik\", \"Product schema\",\n\"BreadcrumbList itemListElement\", \"FAQPage validation\",\n\"Article headline missing\"\n","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).

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.

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.

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:

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.

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.

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)

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)

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

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.

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

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.

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

• TODO/fallback YASAK — every DURUR raises, none silently downgrade (exception: 0-row case is a clean no-op exit, mirrors cannibalization DURUR #7).
• 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).
• Plugin-agnostik — no slug literals; project_slug flows through every path; transform has 0 hardcoded slug words.
• ADR-013: Use when/Also use when/Do not use when are STRING content inside description, not separate fields.
• 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.
• F1: write target is master.xlsx (lowercase, schema-shaped).
• F5: outputs.* values are STRING-TYPED (artifact paths or stringified counts), never raw ints.
• Append-only state — events.jsonl only grows; no in-place rewrite.
• 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).