topical-map

name	topical-map
description	Use when: kullanıcı "pillar", "cluster", "topical map", "konu haritası", "pillar/cluster taxonomy", "topical taxonomy", "konu mimarisi", "site içerik mimarisi", "topic map", "pillar page planı", "içerik mimarisi çıkar" der ya da /pseo-topical-map çağırır. DataForSEO keyword_ideas + related_keywords raw JSON'larını inbox/dfs/ altına persist eder, opsiyonel Phase 7 geo_analysis staging'i konsume ederek pillar/cluster taxonomyye enrich note ekler ve master.xlsx#topical_map sheet'ine yazar (workflow_runner approve gate'ten geçer). Also use when: aktif projenin seed keyword'ü tanımlı; location_code/ language_code project.config.dataforseo'dan config-first çözülür (engine-level ülke varsayılanı YOK); budget pre-flight PASS; cluster-map (Phase 8 W-C1 sibling) F-09 cross-sheet konsume edilecek; new-content-plan için pillar/cluster strüktürü gerekiyor. Do not use when: GSC veri ingestion (gsc-pull), DFS volume lookup (dfs-pull), ham keyword aday havuzu üretimi (content-gaps), kümelenmiş keyword listesi üretimi (cluster-map), URL çakışması (cannibalization) gerekiyor — ayrı discovery / planning / ingestion skill'leri. Master.xlsx yokken çağırma; init-project önce çalışmalı (DURUR #3). Budget aşılmışsa fallback YASAK (DURUR #1).
version	1.0
status	active
category	planning
inputs	{"project_slug":{"type":"string","required":true,"description":"Slug; resolves projects/{slug}/master.xlsx + project.config.json."},"seed_keyword":{"type":"string","required":true,"description":"Seed concept driving the taxonomy. Becomes pillar #1 even when absent from DFS results; embedded in every row's note for traceability."},"location_code":{"type":"integer","required":false,"description":"DFS location_code. NO engine default — resolved config-first from project.config.dataforseo.location_code. Explicit value overrides config. See '## Locale resolution'."},"language_code":{"type":"string","required":false,"description":"DFS language_code. NO engine default — resolved config-first from project.config.dataforseo.language_code (derived from language.content_locale at init). See '## Locale resolution'."},"max_pillars":{"type":"integer","required":false,"default":5,"description":"Maximum pillar rows in the emitted topical_map (default 5)."},"max_clusters_per_pillar":{"type":"integer","required":false,"default":6,"description":"Maximum cluster rows per pillar (default 6)."},"max_supporting_per_cluster":{"type":"integer","required":false,"default":0,"description":"Maximum supporting rows per pillar bucket. Default 0 = OFF (keeps initial topical_map writes lean)."},"default_status":{"type":"string","required":false,"default":"TODO","description":"statusEnum seed for new rows (master-excel.schema.json#/definitions/statusEnum)."},"keyword_cap":{"type":"integer","required":false,"default":5000,"description":"DoS guard: bir MCP yanıtı bu limiti geçerse DURUR (KeywordCapExceededError)."}}
outputs	["master.xlsx#topical_map","outputs/reports/{date}-topical-map.md","events.jsonl","inbox/dfs/{date}-keyword_ideas-topical-{slug}.json","inbox/dfs/{date}-related_keywords-topical-{slug}.json"]
consumes	["init-project:projects/{slug}/master.xlsx","geo-analysis:_state/staging/geo_analysis_geo_signals_{date}_{slug}.json","content-gaps:_state/staging/content_gaps_keyword_ideas_{date}_{slug}.json"]
produces	["cluster-map","new-content-plan","drift-check","monthly-report"]
triggers	{"manual":["/pseo-topical-map"],"natural_language":"\"pillar\", \"cluster\", \"topical map\", \"konu haritası\", \"topical taxonomy\",\n\"pillar/cluster taxonomy\", \"konu mimarisi\", \"site içerik mimarisi\",\n\"topic map\", \"pillar page planı\", \"içerik mimarisi çıkar\"\n","hooks":[]}
mcp_tools	{"required":["mcp__dataforseo__dataforseo_labs_google_keyword_ideas","mcp__dataforseo__dataforseo_labs_google_related_keywords"],"optional":["mcp__dataforseo__keywords_data_google_trends_explore"]}
budget	{"uses_paid_mcp":true,"estimated_credits":2}
autonomy	{"confidence":"MEDIUM","requires_approval":true,"safe_auto_execute":false}

topical-map — planning skill (Phase 8 Wave 1)

10-step protocol. Steps map 1:1 to workflow_runner invocations + the spec §16.5 8-step MCP discipline + §16.8 budget pre-flight. Raw JSON drift recovery is mandatory: every DFS response is dropped into inbox/dfs/ before any transform runs, so a transform bug never costs us the upstream payload (which is paid; re-fetch costs credits).

This skill follows the convention authority of skills/discovery/cannibalization/SKILL.md (Phase 7 Wave 1 — 10-step protocol shape + DURUR + workflow_runner approve gate), skills/discovery/content-gaps/SKILL.md (Phase 7 Wave 2 — DFS staging

• budget pre-flight + DoS guard), and skills/discovery/geo-analysis/SKILL.md (Phase 7 Wave 2 — D-003 helper IMPORT discipline + cross-source consume). The 10-step protocol, raw JSON inbox discipline, budget pre-flight, and provenance event format are reused verbatim — only the domain content (target sheet, transform script, pillar/cluster heuristic, F-09 contract) changes. Deviate only with an ADR.

Inputs (frontmatter contract)

Name	Type	Default	Notes
project_slug	string	—	Required. Resolves projects/{slug}/master.xlsx.
seed_keyword	string	—	Required. Becomes pillar #1; anchors the taxonomy.
location_code	integer	(config)	Config-first from project.config.dataforseo.location_code; no engine default. See Locale resolution.
language_code	string	(config)	Config-first from project.config.dataforseo.language_code. See Locale resolution.
max_pillars	integer	5	Upper bound on pillar rows.
max_clusters_per_pillar	integer	6	Upper bound on cluster rows per pillar.
max_supporting_per_cluster	integer	0	Supporting rows per pillar bucket (0 = OFF).
default_status	string	"TODO"	statusEnum seed for new rows.
keyword_cap	integer	5000	DoS guard per MCP tool.

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

Locale resolution (config-first — NO engine-level country default)

location_code / language_code are resolved config-first from the project's own project.config.dataforseo block — schema-required fields (schemas/project-config.schema.json). There is no engine-level country default: the engine is project-agnostic and must never assume Turkey (or any country). Resolution order:

1. Explicit location_code / language_code input (operator override).
2. project.config.dataforseo.location_code + project.config.dataforseo.language_code.
3. Neither available → DURUR #9 (config locale unresolvable). The skill STOPS before any paid MCP call; it never falls back to a country.

language_code is the DFS short code (tr, en), set in the config at init from language.content_locale (e.g. tr-TR→tr, en-CA→en).

Per-market reference (codes VERIFIED from live project configs)

market	content_locale	location_code	language_code	Note
TR	tr-TR	2792	tr	Turkey (country).
CA	en-CA	20120	en	"Ontario,Canada" — a sub-country location; a coarse country code would be wrong.
NG	en-NG	2566	en	Nigeria (country).

These are the markets currently in the portfolio; every code above is read straight from that project's project.config.dataforseo (the authoritative source — not hardcoded here). For a NEW market, resolve the DFS location_code at init via a serp_locations lookup and persist it into project.config.dataforseo — never guess a country code in skill logic.

TR caveat. TR projects (location 2792) are subject to the same dataforseo-mcp-server@2.8.9 TR-forwarding bug documented in skills/ingestion/dfs-pull/SKILL.md; the imported _normalize_dfs_response tolerates the response shapes, and TR-locale verification stays mandatory for TR projects (do not remove it).

Outputs (artifacts produced)

• projects/{slug}/master.xlsx#topical_map — pillar/cluster/supporting rows (10 cols, schema-locked). Order: pillar row → cluster rows (volume desc) → supporting rows (when enabled).
• projects/{slug}/outputs/reports/{date}-topical-map.md — human-readable summary.
• projects/{slug}/_state/events.jsonl — event_kind=provenance (source.kind=dataforseo_mcp, target_excel_sheet=topical_map).
• projects/{slug}/inbox/dfs/{date}-keyword_ideas-topical-{slug}.json and …-related_keywords-topical-{slug}.json — raw labs payloads (drift recovery; paid).

Schema-locked column tuple (master.xlsx#topical_map, 10 cols)

pillar, cluster, primary_keyword, monthly_volume, data_source,
assigned_url, page_type, status, priority, note

Anchored in schemas/master-excel.schema.json#topical_map.required_columns. The transform's TOPICAL_MAP_COLUMNS constant must equal this tuple byte-for-byte; any drift raises RowSchemaError at row build time AND at emit time (defensive double-check).

Page type vocabulary (free string column, no formal schema enum)

The schema does NOT define an enum for page_type (free string column). The skill commits to a small documented vocabulary so cluster-map / new-content-plan / monthly-report can branch deterministically:

page_type	When applied
pillar	Top-of-funnel anchor concept. Seed keyword always wins this.
cluster	Subtopic sharing ≥1 token with its pillar; volume ≥ min_cluster_volume.
supporting	Long-tail, low-volume, demoted from cluster pool. Optional.

New values require an ADR; the transform raises PageTypeError for any value outside this set. Open Question Q-W-C2-01 documents the lack of a formal schema enum — promotion is a Phase 9+ governance decision.

Invariants (cross-sheet)

F-09 (deferred join) — cluster_keywords.assigned_url ⊆ topical_map.assigned_url

This cluster_keywords ⊆ topical_map join is a DEFERRED design rule (schemas/cross-sheet-invariants.json#/deferred_design_rules → F-09, computed_by=consistency_check, status=deferred). It is NOT enforced today by scripts/validation/validate_invariants.py or the drift-check skill — that cross-sheet join awaits the consistency_check tool (Phase 9+). (Do not confuse it with the active F-09 in schemas/cross-sheet-invariants.json#/rules, which validates a different thing: master_task.task_id uniqueness.) topical-map populates the superset side: every emitted row carries assigned_url="" until cluster-map (W-C1 sibling) or manual edits fill it; this skill guarantees only that its own row writes are F-09-compatible. Downstream cross-sheet enforcement is pending the consistency_check tool.

D-01 — topical_map.pillar ⊆ data/pillars.json

CRITICAL. Pillar values must eventually round-trip through data/pillars.json. In W-C2 the seed→pillar mapping is documented in output but NOT re-keyed automatically (Phase 9 governance). Manager confirms the seed before approving the workflow run.

D-03 — URL idempotency

url_normalizer(x) == x everywhere. No-op for this skill (writes empty assigned_url); D-03 hardening lives in cluster-map / new-content-plan.

D-003 helper IMPORT discipline (mandatory, no copy)

from scripts.ingestion.dfs_pull import _normalize_dfs_response

The function is owned by scripts/ingestion/dfs_pull.py:178 (REST envelope tasks[0].result[0].items AND flat wrapper items tolerance). Copying would silently fork the contract on the next DFS shape change. Identity is locked by tests/skills/test_topical_map.py::test_topical_map_dfs_normalize_helper_imported (is-check). KOPYA YASAK.

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 — preflight_budget (§16.8, MANDATORY for paid-MCP skills)

Subprocess wrapper around scripts/budget/check_budget.py (W-A3/W-A4 paterni). DURUR #1 if exceeded — never silently downgrade.

from scripts.planning import topical_map_transform as tmt

envelope = tmt.preflight_budget(
    project_config_path=project_root / "project.config.json",
    events_path=project_root / "_state" / "events.jsonl",
)
# envelope == {"budget_per_day": 500, "used_24h": 15, "remaining": 485, "exceeded": false}

The wrapper raises BudgetExceededError when the subprocess exits non-zero. Inherits from TopicalMapError so a single except handles every DURUR class this skill emits.

Step 2 — create_run

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

from scripts.state import workflow_runner
handle = workflow_runner.create_run(
    skill="topical-map",
    project_slug=project_slug,
    steps=[
        {"name": "preflight_budget"},
        {"name": "fetch_keyword_ideas"},
        {"name": "fetch_related_keywords"},
        {"name": "consume_geo_staging"},  # OPTIONAL
        {"name": "transform"},
        {"name": "request_approval"},
        {"name": "write_excel"},
        {"name": "render_report"},
    ],
)

Step 3 — fetch_keyword_ideas + Step 4 — fetch_related_keywords (MCP §16.5 step 3 — raw inbox first)

workflow_runner.start_step(handle.run_id, 1, project_slug=project_slug)
# location_code / language_code resolved config-first (see "Locale
# resolution") — no engine-level country default.
raw_ideas = mcp__dataforseo__dataforseo_labs_google_keyword_ideas(
    keywords=[seed_keyword],
    location_code=location_code,    # from project.config.dataforseo
    language_code=language_code,    # from project.config.dataforseo
    limit=keyword_cap,
)
raw_related = mcp__dataforseo__dataforseo_labs_google_related_keywords(
    keyword=seed_keyword,
    location_code=location_code, language_code=language_code,
    depth=2, limit=keyword_cap,
)
# Persist raw → drift recovery (DFS is paid; re-fetch costs credits).
for label, payload in (("keyword_ideas", raw_ideas), ("related_keywords", raw_related)):
    inbox_path = (workspace_root / "projects" / project_slug / "inbox" / "dfs"
                  / f"{today.isoformat()}-{label}-topical-{project_slug}.json")
    inbox_path.parent.mkdir(parents=True, exist_ok=True)
    inbox_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2))

Step 5 — consume_geo_staging (OPTIONAL, Phase 7 W-B4 cross-ref)

When _state/staging/geo_analysis_geo_signals_{date}_{slug}.json exists, load it via topical_map_transform._load_geo_staging(path). Each row's note column is enriched with the matching geo_gap + llm_visibility_score signals. Missing/unreadable file → graceful skip (_load_geo_staging returns []) — DURUR #5 fires ONLY when the caller explicitly required geo-staging via a CLI flag and the file is broken.

from scripts.planning import topical_map_transform as tmt

geo_path = (workspace_root / "projects" / project_slug
            / "_state" / "staging"
            / f"geo_analysis_geo_signals_{today.isoformat()}_{project_slug}.json")
geo_rows = tmt._load_geo_staging(geo_path) if geo_path.exists() else []

Step 6 — transform

Pure compute via scripts/planning/topical_map_transform.py:

python3 scripts/planning/topical_map_transform.py \
    --raw-keyword-ideas    inbox/dfs/{date}-keyword_ideas-topical-{slug}.json \
    --raw-related-keywords inbox/dfs/{date}-related_keywords-topical-{slug}.json \
    --seed-keyword         "{seed}" \
    --geo-staging          _state/staging/geo_analysis_geo_signals_{date}_{slug}.json \
    --max-pillars          5 \
    --max-clusters-per-pillar 6 \
    --default-status       TODO \
    --output-dir           _state/transform/{run_id}/

Produces a JSON object {"topical_map": [<row>, ...], "meta": {...}} shaped to the 10-column master-excel schema. Idempotent: same input → byte-identical output.

Pillar / cluster heuristic (transform domain)

1. Gather DFS candidates from keyword_ideas + related_keywords. Dedupe
   by lower-case keyword; higher-volume entries win on duplicates.
2. PILLAR selection (deterministic):
     - seed_keyword is ALWAYS pillar #1 (synthetic if missing from DFS).
     - Remaining pillar slots fill with candidates that:
         (a) clear min_pillar_volume (default 100),
         (b) share ≥1 token with the seed (sub-pillar of same overall topic),
         (c) don't share their token-set with an already-selected pillar
             (avoids "dental clinic" + "dental clinics" double-pillar).
3. CLUSTER bucketing:
     - For each remaining candidate, compute Jaccard overlap against
       every pillar; assign to the pillar with the maximum overlap > 0.
       Candidates with 0 overlap to all pillars are SKIPPED (sibling
       topics, not in this taxonomy).
4. Per pillar, sort assigned candidates by volume desc, take the first
   max_clusters_per_pillar as cluster rows; below min_cluster_volume
   → demoted to supporting pool.
5. Emit rows: pillar block first (pillar row → cluster rows → supporting
   rows), in pillar order from Step 2.

Geo signal cross-reference (note enrichment)

note = "seed=<seed>; geo_gap=<gap>; llm_vis=<score>; pillar_vol=<n>"

Only fields with signal land in the note string; absent signals collapse to nothing. The note column stays free-form; downstream parsers are optional.

Step 7 — request_approval (skill EXIT awaiting_approval)

workflow_runner.request_approval(
    handle.run_id, project_slug=project_slug,
    approver="user",
    subject=(f"{result['meta']['pillar_count']} pillar / "
             f"{result['meta']['cluster_count']} cluster bulundu, "
             f"master.xlsx#topical_map'e yazalım mı?"),
    step_index=4,
)
# Skill exits here. The user replies in a fresh session; resume below.

Step 8 — Resume (approve → write_excel)

Single committer.commit call — 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 topical_map snapshot sheet). Goes through the single approved write path (backup + lock + schema validation + post-write provenance event).

workflow_runner.approve(handle.run_id, project_slug=project_slug, approver="user")
from scripts.orchestration import committer
committer.commit(
    workspace_root/"projects"/project_slug/"master.xlsx",
    "topical_map",
    topical_map_rows,
    run_id=handle.run_id,
    project_slug=project_slug,
    writer="topical-map",
)

Step 9 — render_report + Provenance event

render_template.py templates/reports/topical-map.template.md data.json → outputs/reports/{date}-topical-map.md. Variables: $project_slug, $date, $seed_keyword, $pillar_count, $cluster_count, $top_pillar, $top_pillar_volume, $top_cluster, $top_cluster_volume, $geo_gaps_summary, $report_summary.

from scripts.state import events_writer
events_writer.append_provenance(
    project_id=project_slug,
    source={"kind": "dataforseo_mcp", "mcp_server": "dataforseo",
            "mcp_tool": "dataforseo__dataforseo_labs_google_keyword_ideas",
            "response_bytes": len(json.dumps(raw_ideas))},
    operation="project_excel",
    target_excel_sheet="topical_map",
    rows_written=len(topical_map_rows),
    cost={"provider": "dataforseo",
          "credits": float(estimated_credits),
          "budget_key": "project.config.dataforseo.budget_credits_per_day"},
)
# Repeat for related_keywords (separate provenance entry per MCP call).

Step 10 — complete

workflow_runner.complete(handle.run_id, project_slug=project_slug, outputs={
    # F5: outputs.* must be STRING-TYPED.
    "pillar_count":   str(result["meta"]["pillar_count"]),
    "cluster_count":  str(result["meta"]["cluster_count"]),
    "row_count":      str(result["meta"]["row_count"]),
    "report_path":    str(report_path),
    "raw_jsons":      ";".join([str(inbox_path), str(related_inbox)]),
    "credits_used":   str(estimated_credits),
})

Workflow status flips running → done (workflow-run schema) and a workflow_action=done event lands in events.jsonl (ADR-020).

DURUR conditions (≥6)

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

1. Budget pre-flight FAIL — scripts/budget/check_budget.py exits non-zero → BudgetExceededError. DFS is paid; running over budget is unrecoverable from this side.
2. DFS payload unparseable — mcp__dataforseo__…_keyword_ideas / …_related_keywords returns auth/network error OR a shape the imported _normalize_dfs_response refuses (ValueError("Unrecognized DFS response shape")). STOP — wrapper drift is an ADR-level event.
3. Column tuple drift — master.xlsx#topical_map column count or names don't match the 10-column schema → RowSchemaError at row build OR emit time.
4. Row-level RowSchemaError — monthly_volume not int, status not in statusEnum, default_status argument invalid.
5. geo-analysis staging missing when required — caller opted into --require-geo-staging but file missing/unreadable → GeoStagingMissingError. Default is graceful skip; this DURUR fires only on opt-in.
6. page_type vocabulary violation — page_type outside {pillar, cluster, supporting} → PageTypeError. New values require an ADR (Q-W-C2-01).
7. keyword_cap exceeded (default 5000) → KeywordCapExceededError. DoS prevention.
8. inbox/dfs/ path unwritable (workspace misconfigured / readonly) or workflow_runner.create_run fails schema validation (schemas/workflow-run.schema.json).
9. Config locale unresolvable — location_code / language_code cannot be resolved (no explicit input AND project.config.dataforseo missing/unreadable). STOP before any paid MCP call; never default to a country (engine is project-agnostic). See Locale resolution.

Cross-references

• Schemas: master-excel.schema.json#topical_map (10 cols + statusEnum), cross-sheet-invariants.json (F-09 + D-01), events.schema.json (dataforseo_mcp / topical_map), skill-frontmatter.schema.json (this frontmatter).
• Cross-modules (IMPORT-only): workflow_runner, transaction, events_writer, check_budget (subprocess), render_template. From dfs_pull: _normalize_dfs_response imported (KOPYA YASAK, identity test).
• Transform: scripts/planning/topical_map_transform.py (transform, estimate_credits, preflight_budget, _load_geo_staging, TOPICAL_MAP_COLUMNS, PAGE_TYPE_*).
• Tests: tests/skills/test_topical_map.py (15 cases incl. import- identity + 10-col tuple lock).
• Phase 8 cluster-map (W-C1 sibling) consumes master.xlsx#topical_map as the F-09 superset before writing master.xlsx#cluster_keywords. The cluster_keywords ⊆ topical_map join is a DEFERRED design rule (computed_by=consistency_check, Phase 9+) — NOT yet enforced by drift-check / scripts/validation/validate_invariants.py.

Discipline checklist

• TODO/fallback YASAK — every DURUR raises, none silently downgrade.
• Schema-first — frontmatter validates against skill-frontmatter.schema.json Draft 7; budget block obeys additionalProperties:false (Q-W-A4-01).
• 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.
• Budget pre-flight at Step 1 BEFORE any paid call.
• Cross-module IMPORT discipline — transaction, workflow_runner, events_writer imported, not modified. _normalize_dfs_response imported from dfs_pull (KOPYA YASAK enforced by identity test).
• DoS guard — keyword_cap default 5000.
• F1: write target is master.xlsx (lowercase, schema-shaped).
• F5: outputs.* STRING-TYPED.
• F-09 superset side: every emitted row carries assigned_url="" until cluster-map (Phase 8 W-C1) or manual edits fill it.
• Append-only state — events.jsonl only grows.