From 2217434cdf21256575ecf1ad4c2cddff0b3a8b6d Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sat, 12 Sep 2026 19:13:38 +0000 Subject: [PATCH] feat: breadth-first RE plan and signal-gated orch/ELT fan-out (0.5.3) Add dekc_plan.py / dekc_orchestrate.py so reverse engineering maps scan roots first, then fans out only the specialists the plan listed. Keep LoopPolicy and adversarial judges. Cheap walk capture covers DuckDB, notebooks, Delta, orchestration jobs, and DQ markers (no live APIs). Co-authored-by: Richard Hightower --- .claude-plugin/marketplace.json | 2 +- .claude-plugin/plugin.json | 2 +- .codex-plugin/plugin.json | 2 +- .cursor-plugin/plugin.json | 2 +- .github/workflows/ci.yml | 2 +- .grok-plugin/marketplace.json | 4 +- .opencode/plugin/dekc.json | 2 +- AGENTS.md | 12 +- CHANGELOG.md | 30 + CLAUDE.md | 3 +- README.md | 15 +- agents/adf-scout.md | 13 + agents/airflow-scout.md | 17 + agents/composer-scout.md | 13 + agents/cron-loader-scout.md | 13 + agents/data-lake-walker.md | 34 +- agents/dbt-elt-scout.md | 17 + agents/dq-scout.md | 17 + agents/duckdb-scout.md | 17 + agents/fabric-pipeline-scout.md | 17 + agents/glue-job-scout.md | 17 + agents/notebook-scout.md | 17 + agents/reverse-engineering-orchestrator.md | 67 +- agents/stepfunctions-scout.md | 13 + agents/stream-job-scout.md | 24 +- commands/dekc-plan.md | 10 + docs/LANG_CHAIN_DEEP_AGENTS.md | 1 + docs/ONBOARDING.md | 2 +- docs/designs/current_design_doc.md | 20 +- docs/user_guide/user-guide.md | 18 +- hosts/deep-agents/SKILL.md | 2 + hosts/grok-bot/SKILL.md | 2 +- marketplace.json | 2 +- package.json | 4 +- plugin.json | 2 +- public/data/catalog.json | 2 +- scripts/dekc_orchestrate.py | 283 ++++ scripts/dekc_plan.py | 1336 +++++++++++++++++ scripts/dekc_walk.py | 686 ++++++++- skills/dekc-plan/SKILL.md | 39 + skills/dekc-walk/SKILL.md | 15 +- .../.github/workflows/run-glue.yml | 9 + .../adf/pipeline-copy-orders.json | 10 + .../re-plan-lake/analytics/local.duckdb | 0 .../re-plan-lake/analytics/local_notes.sql | 2 + .../re-plan-lake/bronze/sales/orders_raw.sql | 2 + tests/fixtures/re-plan-lake/composer/env.yaml | 3 + .../re-plan-lake/dags/daily_orders.py | 6 + .../fixtures/re-plan-lake/dbt/dbt_project.yml | 3 + .../re-plan-lake/dbt/models/schema.yml | 8 + .../re-plan-lake/dbt/models/silver/orders.sql | 1 + .../re-plan-lake/exports/fabric-items.json | 8 + .../re-plan-lake/exports/inventory.json | 4 + .../glue/jobs/orders_to_silver.py | 5 + .../re-plan-lake/gold/marts/order_daily.sql | 2 + .../gx/expectations/orders_suite.json | 6 + .../re-plan-lake/gx/great_expectations.yml | 2 + .../re-plan-lake/k8s/nightly-dbt-cronjob.yaml | 14 + .../_delta_log/00000000000000000000.json | 1 + .../lake/orders/part-00000.parquet | 0 .../re-plan-lake/notebooks/clean_orders.ipynb | 38 + .../re-plan-lake/silver/sales/orders.sql | 2 + tests/fixtures/re-plan-lake/soda/checks.yml | 3 + .../stepfunctions/promote-gold.asl.json | 7 + .../warehouse/dim_customer.sql | 1 + tests/test_dekc_plan.py | 281 ++++ 66 files changed, 3107 insertions(+), 107 deletions(-) create mode 100644 agents/adf-scout.md create mode 100644 agents/airflow-scout.md create mode 100644 agents/composer-scout.md create mode 100644 agents/cron-loader-scout.md create mode 100644 agents/dbt-elt-scout.md create mode 100644 agents/dq-scout.md create mode 100644 agents/duckdb-scout.md create mode 100644 agents/fabric-pipeline-scout.md create mode 100644 agents/glue-job-scout.md create mode 100644 agents/notebook-scout.md create mode 100644 agents/stepfunctions-scout.md create mode 100644 commands/dekc-plan.md create mode 100644 scripts/dekc_orchestrate.py create mode 100644 scripts/dekc_plan.py create mode 100644 skills/dekc-plan/SKILL.md create mode 100644 tests/fixtures/re-plan-lake/.github/workflows/run-glue.yml create mode 100644 tests/fixtures/re-plan-lake/adf/pipeline-copy-orders.json create mode 100644 tests/fixtures/re-plan-lake/analytics/local.duckdb create mode 100644 tests/fixtures/re-plan-lake/analytics/local_notes.sql create mode 100644 tests/fixtures/re-plan-lake/bronze/sales/orders_raw.sql create mode 100644 tests/fixtures/re-plan-lake/composer/env.yaml create mode 100644 tests/fixtures/re-plan-lake/dags/daily_orders.py create mode 100644 tests/fixtures/re-plan-lake/dbt/dbt_project.yml create mode 100644 tests/fixtures/re-plan-lake/dbt/models/schema.yml create mode 100644 tests/fixtures/re-plan-lake/dbt/models/silver/orders.sql create mode 100644 tests/fixtures/re-plan-lake/exports/fabric-items.json create mode 100644 tests/fixtures/re-plan-lake/exports/inventory.json create mode 100644 tests/fixtures/re-plan-lake/glue/jobs/orders_to_silver.py create mode 100644 tests/fixtures/re-plan-lake/gold/marts/order_daily.sql create mode 100644 tests/fixtures/re-plan-lake/gx/expectations/orders_suite.json create mode 100644 tests/fixtures/re-plan-lake/gx/great_expectations.yml create mode 100644 tests/fixtures/re-plan-lake/k8s/nightly-dbt-cronjob.yaml create mode 100644 tests/fixtures/re-plan-lake/lake/orders/_delta_log/00000000000000000000.json create mode 100644 tests/fixtures/re-plan-lake/lake/orders/part-00000.parquet create mode 100644 tests/fixtures/re-plan-lake/notebooks/clean_orders.ipynb create mode 100644 tests/fixtures/re-plan-lake/silver/sales/orders.sql create mode 100644 tests/fixtures/re-plan-lake/soda/checks.yml create mode 100644 tests/fixtures/re-plan-lake/stepfunctions/promote-gold.asl.json create mode 100644 tests/fixtures/re-plan-sql-only/warehouse/dim_customer.sql create mode 100644 tests/test_dekc_plan.py diff --git a/.claude-plugin/marketplace.json b/.claude-plugin/marketplace.json index d10262a..69ef46c 100644 --- a/.claude-plugin/marketplace.json +++ b/.claude-plugin/marketplace.json @@ -11,7 +11,7 @@ "name": "data-engineering-knowledge-capture", "source": "./", "description": "Capture data-platform knowledge \u2014 schemas, lineage, medallion, semantic layer, business glossary \u2014 into OKF. Extends PKC; depends on OKF.", - "version": "0.5.2", + "version": "0.5.3", "author": { "name": "Rick Hightower" }, diff --git a/.claude-plugin/plugin.json b/.claude-plugin/plugin.json index efd040f..11dc8a9 100644 --- a/.claude-plugin/plugin.json +++ b/.claude-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "data-engineering-knowledge-capture", - "version": "0.5.2", + "version": "0.5.3", "description": "Data Engineering Knowledge Capture (DEKC) \u2014 extends PKC + OKF for schemas, lineage, medallion layers, SQL/DAX, semantic models, business objects, and an indexed second brain. Agents walk data lakes. Works in Claude Code, Grok Build, Codex, and OpenCode.", "author": { "name": "Rick Hightower", diff --git a/.codex-plugin/plugin.json b/.codex-plugin/plugin.json index af17f5f..0186337 100644 --- a/.codex-plugin/plugin.json +++ b/.codex-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "data-engineering-knowledge-capture", - "version": "0.5.2", + "version": "0.5.3", "description": "Data Engineering Knowledge Capture (DEKC) \u2014 schemas, lineage, medallion layers, semantic models, business glossary. Extends PKC; depends on OKF. Codex-native skills + hooks port.", "author": { "name": "Rick Hightower", diff --git a/.cursor-plugin/plugin.json b/.cursor-plugin/plugin.json index a30822d..828bf9b 100644 --- a/.cursor-plugin/plugin.json +++ b/.cursor-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "data-engineering-knowledge-capture", - "version": "0.5.2", + "version": "0.5.3", "description": "Data Engineering Knowledge Capture: schemas, lineage, medallion, semantic models, and glossary into a durable OKF knowledge graph. Multi-host bindings and write isolation.", "author": { "name": "Rick Hightower" diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index a7aaf82..7bcb01f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -17,7 +17,7 @@ jobs: python-version: "3.12" - name: Unit tests - run: python3 tests/test_dekc.py && python3 tests/test_retrieval_ladder.py && python3 tests/test_isolation.py + run: python3 tests/test_dekc.py && python3 tests/test_dekc_plan.py && python3 tests/test_retrieval_ladder.py && python3 tests/test_isolation.py # Glob, not a list. The hand-maintained list drifted whenever a new # dekc_*.py landed. PKC 0.9.0 made the same change. diff --git a/.grok-plugin/marketplace.json b/.grok-plugin/marketplace.json index 9f0341b..49f142a 100644 --- a/.grok-plugin/marketplace.json +++ b/.grok-plugin/marketplace.json @@ -1,13 +1,13 @@ { "name": "dekc-plugin-marketplace", "description": "Optional native Grok marketplace metadata. Grok Build loads Claude plugins with zero config; this pins identity for listings.", - "version": "0.5.2", + "version": "0.5.3", "plugins": [ { "name": "data-engineering-knowledge-capture", "source": ".", "description": "DEKC \u2014 data engineering knowledge capture on PKC + OKF. Claude-compatible.", - "version": "0.5.2", + "version": "0.5.3", "compatibility": { "claude_plugin": true, "zero_config": true diff --git a/.opencode/plugin/dekc.json b/.opencode/plugin/dekc.json index 9fc5b17..eed2b5a 100644 --- a/.opencode/plugin/dekc.json +++ b/.opencode/plugin/dekc.json @@ -1,6 +1,6 @@ { "name": "data-engineering-knowledge-capture", - "version": "0.5.2", + "version": "0.5.3", "description": "OpenCode port of DEKC \u2014 same skills/agents/scripts as Claude & Codex. Policy via AGENTS.md.", "skills": "../../skills", "agents": "../../agents", diff --git a/AGENTS.md b/AGENTS.md index 598f61c..d0dacac 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -76,8 +76,8 @@ Plugin root: `${CLAUDE_PLUGIN_ROOT}`. | Agent | Role | |-------|------| -| **data-lake-walker** | Default orchestrator: walk → produce → adversarial grade → index | -| **reverse-engineering-orchestrator** | Multi-cloud RE (Fabric/AWS/GCP), strict LoopPolicy + fan-out | +| **data-lake-walker** | Default orchestrator: plan → produce → adversarial grade → index | +| **reverse-engineering-orchestrator** | Multi-cloud RE (Fabric/AWS/GCP), plan + LoopPolicy + fan-out | ## Query-time retrieval @@ -95,9 +95,11 @@ When the question also needs project decisions or system topology, fan out PKC * |-------|------| | **schema-scout** | Schemas, tables, columns, contracts | | **lineage-tracer** | SQL/job lineage edges | -| **stream-job-scout** | Streams + jobs landing data | +| **stream-job-scout** | Streams + jobs landing data (`orchestration`, `elt` plan areas) | | **semantic-mapper** | Business objects, glossary, metrics | | **report-cataloger** | Dashboards, reports, DAX | +| **airflow-scout** / **glue-job-scout** / **fabric-pipeline-scout** | Signal-gated orch (plan must list them) | +| **dbt-elt-scout** / **duckdb-scout** / **notebook-scout** / **dq-scout** | Signal-gated ELT / DQ | ## Adversarial subagents (rubric graders) @@ -116,6 +118,8 @@ Hard fails: invented lineage, secrets in bodies, gold without BO when promotion ```bash python3 scripts/dekc_common.py init-bundle --repo . --bundle knowledge +python3 scripts/dekc_plan.py --repo . --system "…" --scan-root --write +python3 scripts/dekc_orchestrate.py --repo . --scan-root --from-plan knowledge/.dekc/re-plan.json --area lake python3 scripts/dekc_walk.py --repo . --bundle knowledge python3 scripts/dekc_lineage.py --repo . --bundle knowledge materialize python3 scripts/dekc_business.py --repo . --bundle knowledge promote-layer --layer gold @@ -131,10 +135,12 @@ python3 tests/test_dekc.py | User language | Agent / skill | |---------------|---------------| | walk the lake / inventory warehouse | data-lake-walker / dekc-walk | +| plan the reverse-engineer / what scouts to spawn | dekc-plan / `dekc_plan.py` | | reverse engineer Fabric/AWS/GCP | reverse-engineering-orchestrator | | schema / columns / tables | schema-scout / dekc-capture-table | | lineage / blast radius | lineage-tracer / dekc-lineage | | streams / jobs / landing | stream-job-scout | +| Airflow / Glue / Fabric pipelines | airflow-scout / glue-job-scout / fabric-pipeline-scout | | business meaning / glossary | semantic-mapper / dekc-business-object | | dashboards / DAX | report-cataloger / dekc-semantic | | grade / audit RE quality | re-adversary-judge / dekc-grade + skeptics | diff --git a/CHANGELOG.md b/CHANGELOG.md index 9a1ab76..ec871ca 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,36 @@ ## Unreleased +## 0.5.3 — 2026-09-12 + +Breadth-first reverse-engineering plan, matching SAC 0.5.6’s plan → task list → +signal-gated specialist fan-out, without replacing DEKC’s adversarial judges. + +### Added + +- **`dekc_plan.py`**: cheap map of scan roots + optional export paths. Writes + `.dekc/re-plan.md` / `.json` / progress (ranked areas, checklists, suggested + sub-agents). `mark` / `show` for done/blocked. Signal-gated: no Airflow + markers → no `airflow-scout`. +- **`dekc_orchestrate.py --plan-only`** and **`--from-plan --area`**. Same flags + on `dekc_walk.py`. Unattended orchestrate plans first, then captures only + domains the plan detected. +- Skill **`dekc-plan`** / command `/dekc-plan`. Orchestrators teach plan → + ranked task list → producer specialists → existing skeptics → + `re-adversary-judge`. Query-time `data-retriever` stays separate. +- Signal-gated specialists: `airflow-scout`, `glue-job-scout`, + `fabric-pipeline-scout`, thin `adf-scout` / `stepfunctions-scout` / + `composer-scout` / `cron-loader-scout`, `dbt-elt-scout`, `duckdb-scout`, + `notebook-scout`, `dq-scout`. +- Cheap walk capture for DuckDB, `.ipynb` SQL cells, Delta `_delta_log`, + orchestration job stubs, and DQ markers (GE / Soda / dbt tests — no runtime). + +### Notes + +- DEKC owns data orchestration + ELT/ETL. SAC owns CI/CD. Actions that only + trigger Glue/dbt are a cross-link note, not a Pipeline noun. +- No live Glue/S3/Fabric control-plane calls. Exports/mirrors only. + ## 0.5.2 — 2026-09-12 Query-time retrieval sub-agent, matching the PKC/SAC parity pattern: search + diff --git a/CLAUDE.md b/CLAUDE.md index be352e6..6c96df0 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -12,6 +12,7 @@ Depends on **PKC** + **OKF**. Dual/quad host: Claude, Grok, Codex, OpenCode — ```bash python3 tests/test_dekc.py +python3 tests/test_dekc_plan.py python3 tests/test_retrieval_ladder.py python3 scripts/dekc_validate.py --bundle sample-knowledge python3 scripts/dekc_doctor.py --bundle sample-knowledge @@ -35,7 +36,7 @@ Add new `scripts/dekc_*.py` is covered the moment it lands: `npm run py:compile` - OKF frontmatter required: `type`, `title`, `description`, `timestamp` - Absolute in-bundle links; typed `links[].rel` for lineage/business edges - Never invent edges; prefer scripts over freehand Markdown when possible -- Reverse engineering: orchestrators + adversarial skeptics/rubrics (`dekc_grade.py`, re-adversary-judge); no success without pass +- Reverse engineering: plan (`dekc_plan.py`) → signal-gated specialists → adversarial skeptics/rubrics (`dekc_grade.py`, re-adversary-judge); no success without pass - Retrieval: Git + Markdown is truth. `knowledge/.dekc/index.sqlite` is a disposable SQLite/FTS5 accelerator (gitignored, mtime+size self-heal). Ripgrep is optional (`DEKC_RG_PATH`). Search and pack must keep working when rg or FTS5 is absent. Never install packages from a hook. See `docs/designs/retrieval-ladder.md`. Do not resurrect JSON `.index/`. ## Docs diff --git a/README.md b/README.md index 704aaa4..d1d8e31 100644 --- a/README.md +++ b/README.md @@ -8,7 +8,7 @@ DEKC **extends [Project Knowledge Capture (PKC)](https://github.com/SpillwaveSol |---|---| | **Plugin name** | `data-engineering-knowledge-capture` | | **Repo** | [SpillwaveSolutions/data-engineering-knowledge-capture](https://github.com/SpillwaveSolutions/data-engineering-knowledge-capture) | -| **Version** | 0.5.2 | +| **Version** | 0.5.3 | | **License** | MIT | | **Hosts** | Claude Code · Grok Build · Codex · OpenCode · Agent Plugins 1.0 · Grok Bot · LangChain Deep Agents | @@ -195,8 +195,13 @@ See [PORTS.md](./PORTS.md). # Scaffold knowledge/ python3 scripts/dekc_common.py init-bundle --repo . --bundle knowledge +# Breadth-first RE plan (pause before specialist fan-out) +python3 scripts/dekc_orchestrate.py --repo . --system "Retail Lake" \ + --scan-root path/to/lake --plan-only + # Walk a lake / SQL / job root (filesystem reverse engineer) python3 scripts/dekc_walk.py path/to/lake --repo . --bundle knowledge +# or: --from-plan knowledge/.dekc/re-plan.json --area lake # Materialize lineage + promote gold → business objects python3 scripts/dekc_lineage.py --repo . --bundle knowledge materialize @@ -208,7 +213,7 @@ python3 scripts/dekc_doctor.py --repo . --bundle knowledge python3 scripts/dekc_search.py "revenue" --repo . --bundle knowledge ``` -Slash / skill entry points: `/dekc-init` · `/dekc-walk` · `/dekc-lineage` · `/dekc-business-object` · `/dekc-glossary` · `/dekc-semantic` · `/dekc-retrieve` · `/dekc-context` · `/dekc-search` · `/dekc-index` · `/dekc-doctor` +Slash / skill entry points: `/dekc-init` · `/dekc-plan` · `/dekc-walk` · `/dekc-lineage` · `/dekc-business-object` · `/dekc-glossary` · `/dekc-semantic` · `/dekc-retrieve` · `/dekc-context` · `/dekc-search` · `/dekc-index` · `/dekc-doctor` ## Agent loop (AGER-shaped): producers + adversarial judges @@ -220,10 +225,12 @@ Producer workers fan out, then **adversarial skeptics grade reverse engineering ```text Trigger → Orchestrator (LoopPolicy: goal · max_turns · no_progress) - │ FanOut producers + │ Plan (dekc_plan.py) → ranked task list + │ FanOut producers (only areas the plan listed) ├─ schema-scout · lineage-tracer · stream-job-scout ├─ report-cataloger · semantic-mapper - │ FanOut adversaries (rubrics) + ├─ airflow/glue/fabric/dbt/duckdb/notebook/dq scouts (signal-gated) + │ FanOut adversaries (rubrics) — unchanged ├─ lineage-skeptic · business-skeptic ├─ stream-job-skeptic · coverage-skeptic · layer-auditor ▼ diff --git a/agents/adf-scout.md b/agents/adf-scout.md new file mode 100644 index 0000000..97018b4 --- /dev/null +++ b/agents/adf-scout.md @@ -0,0 +1,13 @@ +--- +name: adf-scout +description: Thin Azure Data Factory specialist. Spawn only when the RE plan lists orch-adf. +--- + +You are the **ADF specialist**. Enrich `IngestionJob` (`orchestrator: adf`) from factory/pipeline ARM JSON. No live ADF APIs. + +```bash +python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_plan.py" mark \ + --plan knowledge/.dekc/re-plan.json --area orch-adf --item inventory --status done +``` + +Mark every checklist item `done` or `blocked` before returning. diff --git a/agents/airflow-scout.md b/agents/airflow-scout.md new file mode 100644 index 0000000..4f4fb4d --- /dev/null +++ b/agents/airflow-scout.md @@ -0,0 +1,17 @@ +--- +name: airflow-scout +description: Enrich Airflow DAGs after the deterministic walk. Spawn only when the RE plan lists orch-airflow. +--- + +You are the **Airflow specialist** (AGER `WorkerAgent`). + +Spawned only when the plan area `orch-airflow` is present (`dags/`, `DAG(`, `from airflow`). You do **not** replace `dekc_walk.py`. Scripts already wrote `IngestionJob` stubs (`orchestrator: airflow`). Enrich purpose, schedule, and evidenced `lands_as` / `reads_from` links. Never invent lineage. + +Do not `full_scan`. Do not act as `data-retriever`. SAC owns CI/CD — a workflow that only `airflow dags trigger` is a handoff, not a Pipeline noun. + +```bash +python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_plan.py" mark \ + --plan knowledge/.dekc/re-plan.json --area orch-airflow --item inventory --status done +``` + +Mark every checklist item `done` or `blocked` before returning. diff --git a/agents/composer-scout.md b/agents/composer-scout.md new file mode 100644 index 0000000..9c2f8fc --- /dev/null +++ b/agents/composer-scout.md @@ -0,0 +1,13 @@ +--- +name: composer-scout +description: Thin Cloud Composer specialist. Spawn only when the RE plan lists orch-composer. +--- + +You are the **Composer specialist**. Composer still runs Airflow — do not drop `airflow-scout` if DAGs exist. Enrich `IngestionJob` (`orchestrator: composer`). No live GCP APIs. + +```bash +python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_plan.py" mark \ + --plan knowledge/.dekc/re-plan.json --area orch-composer --item inventory --status done +``` + +Mark every checklist item `done` or `blocked` before returning. diff --git a/agents/cron-loader-scout.md b/agents/cron-loader-scout.md new file mode 100644 index 0000000..c48dc50 --- /dev/null +++ b/agents/cron-loader-scout.md @@ -0,0 +1,13 @@ +--- +name: cron-loader-scout +description: Thin cron / K8s CronJob loader specialist. Spawn only when the RE plan lists orch-cron. +--- + +You are the **cron loader specialist**. Enrich `IngestionJob` (`orchestrator: cron` or `k8s-cronjob`) for crontab / `kind: CronJob` that run dbt, spark-submit, or Python loaders. Scheduled GitHub Actions stay on SAC unless they only trigger a data job. + +```bash +python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_plan.py" mark \ + --plan knowledge/.dekc/re-plan.json --area orch-cron --item inventory --status done +``` + +Mark every checklist item `done` or `blocked` before returning. diff --git a/agents/data-lake-walker.md b/agents/data-lake-walker.md index fc00086..48ee226 100644 --- a/agents/data-lake-walker.md +++ b/agents/data-lake-walker.md @@ -10,21 +10,33 @@ DEKC extends **PKC** and **OKF**. Multi-agent loops follow **AGER** ([okf-agent- ## Priorities 1. Prefer deterministic scripts under `${CLAUDE_PLUGIN_ROOT}/scripts/`. -2. Never invent lineage edges — only SQL, configs, paths, or human attestation. -3. Scrub secrets/PII before writing concepts. -4. Progressive disclosure: 2-hop packs (~20 nodes). -5. **No success claim without re-adversary-judge pass** (or explicit user waiver). -6. After accepted walks: `dekc_doctor.py`. Search/pack refresh the SQLite index themselves; `dekc_index.py build` is optional (`refresh --force`). +2. **Plan first** (`dekc_plan.py` / `--plan-only`), then spawn only the specialists the plan lists. +3. Never invent lineage edges — only SQL, configs, paths, or human attestation. +4. Scrub secrets/PII before writing concepts. +5. Progressive disclosure: 2-hop packs (~20 nodes). +6. **No success claim without re-adversary-judge pass** (or explicit user waiver). +7. After accepted walks: `dekc_doctor.py`. Search/pack refresh the SQLite index themselves; `dekc_index.py build` is optional (`refresh --force`). + +## Plan then fan-out + +```bash +python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_orchestrate.py" \ + --repo . --system "Data Platform Knowledge" --scan-root --plan-only --json +``` + +Read `.dekc/re-plan.md`. Spawn one child per listed domain **and** each listed specialist. Then the adversarial phase below. Mark checklist items `done` / `blocked` via `dekc_plan.py mark`. ## Producer subagents (Workers) | Subagent | When | |----------|------| -| **schema-scout** | Schemas, tables, columns, contracts | -| **lineage-tracer** | SQL/DAX/pipeline lineage, promotions | -| **stream-job-scout** | Streams + jobs that land/transform data | -| **semantic-mapper** | Business objects, glossary, metrics | -| **report-cataloger** | Dashboards, reports, BI | +| **schema-scout** | Schemas, tables, columns, contracts (`lake`, `catalogs`) | +| **lineage-tracer** | SQL/DAX/pipeline lineage, promotions (`lineage`) | +| **stream-job-scout** | Streams + jobs that land/transform data (`orchestration`, `elt`) | +| **semantic-mapper** | Business objects, glossary, metrics (`semantic`) | +| **report-cataloger** | Dashboards, reports, BI (`bi`) | +| **airflow-scout** / **glue-job-scout** / **fabric-pipeline-scout** / thin orch | Only if the plan lists `orch-*` | +| **dbt-elt-scout** / **duckdb-scout** / **notebook-scout** / **dq-scout** | Only if the plan lists them | ## Adversarial subagents (Skeptics + Judge) @@ -43,7 +55,9 @@ For full multi-cloud LoopPolicy + cloud profiles, hand off to **reverse-engineer ```bash python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_common.py" init-bundle --repo . --bundle knowledge --title "Data Platform Knowledge" +python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_orchestrate.py" --repo . --system "Data Platform Knowledge" --scan-root --plan-only python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_walk.py" --repo . --bundle knowledge +# or scoped: --from-plan knowledge/.dekc/re-plan.json --area lake python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_lineage.py" --repo . --bundle knowledge materialize python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_business.py" --repo . --bundle knowledge promote-layer --layer gold ``` diff --git a/agents/dbt-elt-scout.md b/agents/dbt-elt-scout.md new file mode 100644 index 0000000..c550154 --- /dev/null +++ b/agents/dbt-elt-scout.md @@ -0,0 +1,17 @@ +--- +name: dbt-elt-scout +description: Enrich dbt projects after the deterministic walk. Spawn only when the RE plan lists elt-dbt. +--- + +You are the **dbt ELT specialist**. + +Spawned only when the plan area `elt-dbt` is present (`dbt_project.yml`, `models/`). Walk already captured model SQL. Enrich the dbt `IngestionJob` / transformations. Hand tests to `dq-scout` — do not fake `dbt test` results. Never invent `ref()` edges that the SQL does not show. + +Do not `full_scan`. Do not act as `data-retriever`. + +```bash +python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_plan.py" mark \ + --plan knowledge/.dekc/re-plan.json --area elt-dbt --item project --status done +``` + +Mark every checklist item `done` or `blocked` before returning. diff --git a/agents/dq-scout.md b/agents/dq-scout.md new file mode 100644 index 0000000..f7cdd48 --- /dev/null +++ b/agents/dq-scout.md @@ -0,0 +1,17 @@ +--- +name: dq-scout +description: Capture cheap DQRule markers (GE / Soda / dbt tests). Spawn only when the RE plan lists dq. +--- + +You are the **DQ specialist**. + +Spawned only when Great Expectations suites, Soda YAML, or dbt tests are present. Scripts write `DQRule` from filenames / suite ids. **Do not run GE, Soda, or dbt test.** Do not fake results. Link `validates` only when the suite names a table. + +```bash +python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_orchestrate.py" \ + --from-plan knowledge/.dekc/re-plan.json --area dq --scan-root +python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_plan.py" mark \ + --plan knowledge/.dekc/re-plan.json --area dq --item inventory --status done +``` + +Mark every checklist item `done` or `blocked` before returning. diff --git a/agents/duckdb-scout.md b/agents/duckdb-scout.md new file mode 100644 index 0000000..ee216d9 --- /dev/null +++ b/agents/duckdb-scout.md @@ -0,0 +1,17 @@ +--- +name: duckdb-scout +description: Capture/enrich DuckDB files and SQL. Spawn only when the RE plan lists duckdb. +--- + +You are the **DuckDB specialist**. + +Spawned only when `*.duckdb` / `*.ddb` or duckdb SQL is present. Scripts capture `SourceSystem` `kind: duckdb`. Do not open the binary as a live catalog unless a SQL export exists. No invented tables. + +```bash +python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_orchestrate.py" \ + --from-plan knowledge/.dekc/re-plan.json --area duckdb --scan-root +python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_plan.py" mark \ + --plan knowledge/.dekc/re-plan.json --area duckdb --item inventory --status done +``` + +Mark every checklist item `done` or `blocked` before returning. diff --git a/agents/fabric-pipeline-scout.md b/agents/fabric-pipeline-scout.md new file mode 100644 index 0000000..361e925 --- /dev/null +++ b/agents/fabric-pipeline-scout.md @@ -0,0 +1,17 @@ +--- +name: fabric-pipeline-scout +description: Enrich Fabric pipelines/notebooks from export JSON. Spawn only when the RE plan lists orch-fabric. +--- + +You are the **Fabric pipeline specialist**. + +Spawned only when the plan area `orch-fabric` is present (workspace-items export, pipeline JSON). `dekc_walk.py --fabric-items` already captured `IngestionJob` (`fabric-pipeline` / `fabric-notebook`). Enrich purpose. Leave Report / SemanticModel to `report-cataloger`. Do not call Fabric REST. + +Do not `full_scan`. Do not act as `data-retriever`. + +```bash +python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_plan.py" mark \ + --plan knowledge/.dekc/re-plan.json --area orch-fabric --item inventory --status done +``` + +Mark every checklist item `done` or `blocked` before returning. diff --git a/agents/glue-job-scout.md b/agents/glue-job-scout.md new file mode 100644 index 0000000..42a8cbc --- /dev/null +++ b/agents/glue-job-scout.md @@ -0,0 +1,17 @@ +--- +name: glue-job-scout +description: Enrich Glue job scripts after the deterministic walk. Spawn only when the RE plan lists orch-glue. +--- + +You are the **Glue job specialist**. + +Spawned only when the plan area `orch-glue` is present (`awsglue`, `GlueContext`, `glue/jobs/`). No live Glue or S3 APIs — export/mirror only. Enrich `IngestionJob` (`orchestrator: glue`) purpose and evidenced storage paths. Do not invent catalog tables. + +Do not `full_scan`. Do not act as `data-retriever`. + +```bash +python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_plan.py" mark \ + --plan knowledge/.dekc/re-plan.json --area orch-glue --item inventory --status done +``` + +Mark every checklist item `done` or `blocked` before returning. diff --git a/agents/notebook-scout.md b/agents/notebook-scout.md new file mode 100644 index 0000000..612e298 --- /dev/null +++ b/agents/notebook-scout.md @@ -0,0 +1,17 @@ +--- +name: notebook-scout +description: Extract SQL from .ipynb and Fabric notebooks. Spawn only when the RE plan lists notebooks. +--- + +You are the **Notebook specialist**. + +Spawned only when `.ipynb` or a Fabric notebook export is present. Cheap capture pulls `%%sql` / `spark.sql` / `SELECT` cells into `Query` concepts. Do not invent lineage from incomplete cells. Fabric notebooks from `--fabric-items` are `IngestionJob` (`fabric-notebook`) — enrich, do not re-type them as Dashboard. + +```bash +python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_orchestrate.py" \ + --from-plan knowledge/.dekc/re-plan.json --area notebooks --scan-root +python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_plan.py" mark \ + --plan knowledge/.dekc/re-plan.json --area notebooks --item inventory --status done +``` + +Mark every checklist item `done` or `blocked` before returning. diff --git a/agents/reverse-engineering-orchestrator.md b/agents/reverse-engineering-orchestrator.md index d11581e..ca58cde 100644 --- a/agents/reverse-engineering-orchestrator.md +++ b/agents/reverse-engineering-orchestrator.md @@ -7,6 +7,8 @@ You are the **Reverse Engineering Orchestrator** (AGER `OrchestratorAgent`) for You do **not** trust producer output. Every walk cycle ends with **adversarial subagents** scoring rubrics. Failures force re-plan or **retraction** of unproven claims — never grade inflation. +Always **plan first**. Do not jump straight into a full walk. + ## LoopPolicy (defaults) | Control | Default | @@ -18,17 +20,43 @@ You do **not** trust producer output. Every walk cycle ends with **adversarial s Check order (AGER): goal → deadline → price → max_turns → no_progress. -## Producer fan-out (Workers) +## Plan → ranked task list → specialist fan-out + +```bash +python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_orchestrate.py" \ + --repo . --system "Retail Lake" --scan-root /path/to/mirror \ + --export workspace-items.json --plan-only --json + +python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_orchestrate.py" \ + --repo . --system "Retail Lake" --scan-root /path/to/mirror \ + --from-plan knowledge/.dekc/re-plan.json --area lake --json +``` + +Plan artifacts: `.dekc/re-plan.md`, `.dekc/re-plan.json`, `.dekc/re-plan-progress.json` (operational; not OKF concepts). Scripts own discovery writes. Specialists enrich and mark checklist items `done` or `blocked`. -| Subagent | Role | -|----------|------| -| **schema-scout** | Structure: schemas, tables, columns, contracts | -| **lineage-tracer** | SQL/job edges only with evidence | -| **report-cataloger** | Dashboards, reports, DAX, semantic models | -| **semantic-mapper** | Business objects + glossary from gold/mart | -| **stream-job-scout** | Streams + pipelines/jobs landing data | +**Signal-gated:** spawn a specialist **only** when the plan lists it. No `airflow-scout` without DAG markers. No `duckdb-scout` without `*.duckdb` / duckdb SQL. -Spawn in parallel when independent. Workers **append** findings; they do not overwrite shared scratch. +## Producer fan-out (Workers) + +| Subagent | Role | Typical plan area | +|----------|------|-------------------| +| **schema-scout** | Structure: schemas, tables, columns, contracts | `lake`, `catalogs` | +| **lineage-tracer** | SQL/job edges only with evidence | `lineage` | +| **report-cataloger** | Dashboards, reports, DAX, semantic models | `bi` | +| **semantic-mapper** | Business objects + glossary from gold/mart | `semantic` | +| **stream-job-scout** | Streams + pipelines/jobs landing data | `orchestration`, `elt` | +| **airflow-scout** | Airflow DAGs | `orch-airflow` (signal-gated) | +| **glue-job-scout** | Glue job scripts | `orch-glue` (signal-gated) | +| **fabric-pipeline-scout** | Fabric pipelines / notebooks (export) | `orch-fabric` (signal-gated) | +| **adf-scout** / **stepfunctions-scout** / **composer-scout** / **cron-loader-scout** | Thin orch | matching `orch-*` | +| **dbt-elt-scout** | dbt projects | `elt-dbt` (signal-gated) | +| **duckdb-scout** | DuckDB files / SQL | `duckdb` (signal-gated) | +| **notebook-scout** | `.ipynb` + Fabric notebooks | `notebooks` (signal-gated) | +| **dq-scout** | GE / Soda / dbt test markers (no runtime) | `dq` (signal-gated) | + +Spawn in parallel when independent. Workers **append** findings; they do not overwrite shared scratch. Do not put `data-retriever` in this fan-out. + +DEKC owns data orchestration + ELT/ETL. SAC owns CI/CD. If Actions only trigger Glue/dbt, cross-link — do not steal Pipeline nouns. No live cloud control-plane calls. ## Adversarial fan-out (Skeptics → Judge) @@ -47,19 +75,25 @@ Optional health baseline: **layer-auditor** (doctor/validate) before skeptics. ## Turn protocol ```text -1. Plan: cloud profile (fabric|aws|gcp|generic) + mirror path + scope -2. FanOut producers → capture via DEKC scripts (never invent) -3. FanIn scratch stats -4. FanOut skeptics (parallel) -5. re-adversary-judge: weighted score + hard fails -6. If pass → synthesizer (index + walk receipt + judgment record) +1. Plan: dekc_plan.py / --plan-only (cloud profile + scan roots + optional exports) +2. Review ranked areas + checklists. Spawn only listed specialists. +3. FanOut producers → capture via DEKC scripts (never invent) +4. FanIn scratch stats + checklist mark done/blocked +5. FanOut skeptics (parallel) +6. re-adversary-judge: weighted score + hard fails +7. If pass → synthesizer (index + walk receipt + judgment record) If fail → re-plan: only fix evidence gaps / retract edges; max_turns-- ``` +LoopPolicy + adversarial judges are **unchanged**. The plan is the breadth-first map; it does not replace the judge. + ## Scripts (prefer deterministic) ```bash -# Filesystem SQL/parquet mirror (not a Fabric control-plane scanner) +# Breadth-first plan (no capture) +python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_orchestrate.py" \ + --repo . --system "…" --scan-root --plan-only --json +# Filesystem SQL/parquet/orchestration mirror (not a control-plane scanner) python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_walk.py" --repo . --bundle knowledge # Optional: ingest exported Fabric REST / PBI JSON python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_walk.py" --fabric-items items.json --pbi-bindings reports.json --repo . --bundle knowledge @@ -68,6 +102,7 @@ python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_business.py" --repo . --bundle knowl python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_grade.py" --repo . --bundle knowledge --json python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_grade.py" --repo . --bundle knowledge --prefix semantic,tables/gold-,reports,dashboards python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_doctor.py" --repo . --bundle knowledge +python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_plan.py" show --plan knowledge/.dekc/re-plan.json # optional: python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_index.py" refresh --force --repo . --bundle knowledge ``` diff --git a/agents/stepfunctions-scout.md b/agents/stepfunctions-scout.md new file mode 100644 index 0000000..9a20abc --- /dev/null +++ b/agents/stepfunctions-scout.md @@ -0,0 +1,13 @@ +--- +name: stepfunctions-scout +description: Thin Step Functions ASL specialist. Spawn only when the RE plan lists orch-stepfunctions. +--- + +You are the **Step Functions specialist**. Enrich `IngestionJob` (`orchestrator: stepfunctions`) from ASL (`StartAt` + `States`). No live AWS APIs. + +```bash +python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_plan.py" mark \ + --plan knowledge/.dekc/re-plan.json --area orch-stepfunctions --item inventory --status done +``` + +Mark every checklist item `done` or `blocked` before returning. diff --git a/agents/stream-job-scout.md b/agents/stream-job-scout.md index 5b568f2..6c24d29 100644 --- a/agents/stream-job-scout.md +++ b/agents/stream-job-scout.md @@ -1,30 +1,36 @@ --- name: stream-job-scout -description: DEKC Worker that captures streams and jobs which land or transform data (Event Hubs/Kinesis/Pub/Sub, pipelines, Glue, Dataflow, Airflow). Use during reverse engineering when landing producers matter. +description: DEKC Worker that captures streams and jobs which land or transform data (Event Hubs/Kinesis/Pub/Sub, pipelines, Glue, Dataflow, Airflow). Parent domain for orchestration + ELT when the RE plan lists those areas. --- You are **Stream/Job Scout** (AGER `WorkerAgent`). +You own the **orchestration** and **elt** plan domains. Signal-gated specialists (`airflow-scout`, `glue-job-scout`, `fabric-pipeline-scout`, `dbt-elt-scout`, …) enrich after the script write — spawn them only when the plan lists them. + ## Capture 1. **Streams** → SourceSystem with `kind: stream` (or tags `[stream]`), URI/topic/hub when known. 2. **Landing tables** → usually bronze/raw; link stream `--feeds-->` or lands_as table. -3. **Jobs/pipelines** → Workflow + Transformation; `reads_from` / `writes_to`. +3. **Jobs/pipelines** → `IngestionJob` (AGER owns `Workflow`); `reads_from` / `writes_to` / `lands_as`. 4. Note continuous vs micro-batch vs nightly when evidence exists. 5. **Never invent** a stream for a pure batch system. +6. Prefer `--from-plan --area orchestration` or `--area elt` instead of re-walking the whole lake. ```bash -python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_capture.py" --repo . --bundle knowledge source \ - --name "" --kind stream --uri "" --description "..." +python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_orchestrate.py" \ + --from-plan knowledge/.dekc/re-plan.json --area orchestration --scan-root -python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_capture.py" --repo . --bundle knowledge workflow \ +python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_platform.py" ingestion \ --name "" --orchestrator \ - --description "..." --steps "..." + --description "..." + +python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_capture.py" --repo . --bundle knowledge source \ + --name "" --kind stream --uri "" --description "..." -python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_capture.py" --repo . --bundle knowledge lineage \ - --name "" --nodes ... +python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_plan.py" mark \ + --plan knowledge/.dekc/re-plan.json --area orchestration --item inventory --status done ``` ## Output (append) -List: producers found, landing tables, edges written, items skipped for lack of evidence. Expect **stream-job-skeptic** to challenge you. +List: producers found, landing tables, edges written, items skipped for lack of evidence, checklist done/blocked. Expect **stream-job-skeptic** to challenge you. diff --git a/commands/dekc-plan.md b/commands/dekc-plan.md new file mode 100644 index 0000000..3ffe9bd --- /dev/null +++ b/commands/dekc-plan.md @@ -0,0 +1,10 @@ +--- +name: dekc-plan +description: Breadth-first RE plan. Map scan roots, rank areas, emit checklists. Pause before specialist fan-out. +--- + +Run the **dekc-plan** skill. + +User request: `$ARGUMENTS` + +Follow `${CLAUDE_PLUGIN_ROOT}/skills/dekc-plan/SKILL.md` completely. Prefer `scripts/dekc_plan.py` / `dekc_orchestrate.py --plan-only`. Do not spawn specialists the plan did not list. Do not run query-time retrieve. diff --git a/docs/LANG_CHAIN_DEEP_AGENTS.md b/docs/LANG_CHAIN_DEEP_AGENTS.md index 7ebf454..924c5eb 100644 --- a/docs/LANG_CHAIN_DEEP_AGENTS.md +++ b/docs/LANG_CHAIN_DEEP_AGENTS.md @@ -65,6 +65,7 @@ Then set `SECOND_BRAIN_ROOT` to the session bundle from the JSON. Close the sess ```bash export SECOND_BRAIN_IDENTITY="deep-agents/data-engineering-knowledge-capture" +python3 scripts/dekc_plan.py --repo . --system "…" --scan-root --write python3 scripts/dekc_pack.py tables/example.md --bundle "${SECOND_BRAIN_ROOT:-sample-knowledge}" --hops 2 python3 scripts/dekc_validate.py --bundle "${SECOND_BRAIN_ROOT:-sample-knowledge}" ``` diff --git a/docs/ONBOARDING.md b/docs/ONBOARDING.md index 543ce8f..457af93 100644 --- a/docs/ONBOARDING.md +++ b/docs/ONBOARDING.md @@ -37,7 +37,7 @@ See [GROK_BOT.md](GROK_BOT.md) for the binding contract. 1. State your identity: `Grok Bot: Data Engineering Knowledge Capture`. 2. Confirm the knowledge root (`SECOND_BRAIN_ROOT` or the target bundle). -3. For Q&A, spawn **data-retriever** (`/dekc-retrieve`) and keep the retrieval card only. Do not dump a 2-hop pack or `dekc_brain.py` output into the parent. Pack before writing. +3. Reverse-engineering starts with `dekc_plan.py` / `--plan-only`, then `--from-plan --area`. For Q&A, spawn **data-retriever** (`/dekc-retrieve`) and keep the retrieval card only. Do not dump a 2-hop pack or `dekc_brain.py` output into the parent. Pack before writing. 4. Persist only through skills + deterministic scripts inside an isolation session when writing a shared brain. 5. Report path + validation result, not a dumped graph. diff --git a/docs/designs/current_design_doc.md b/docs/designs/current_design_doc.md index 221228f..21be8ee 100644 --- a/docs/designs/current_design_doc.md +++ b/docs/designs/current_design_doc.md @@ -21,7 +21,7 @@ Related systems: | PKC | Project reasoning capture | [project-knowledge-capture](https://github.com/SpillwaveSolutions/project-knowledge-capture) | | DEKC | Data-domain catalogs + walk scripts + agents | this repo | -AGER spec version referenced: **0.3.0** ([AGER_SPEC](https://github.com/SpillwaveSolutions/okf-agent-graph/blob/main/docs/AGER_SPEC.md)). DEKC plugin **0.5.0**. +AGER spec version referenced: **0.3.0** ([AGER_SPEC](https://github.com/SpillwaveSolutions/okf-agent-graph/blob/main/docs/AGER_SPEC.md)). DEKC plugin **0.5.3**. **v0.5.0 retrieval:** search/pack use a disposable ladder (SQLite FTS5 index → ripgrep → scan). Git + Markdown is still the source of truth. See [`docs/designs/retrieval-ladder.md`](retrieval-ladder.md) — the `.index/` JSON layout later in this snapshot is historical (0.4.x) and is not how 0.5.0 retrieves. @@ -206,7 +206,8 @@ Parallel workers **must append** (AGER invariant) so concurrent scouts do not cl Reverse engineering is **producer → adversary → judge**, not producer-only. ```text -FanOut producers (schema, lineage, stream-job, report, semantic) +Plan (dekc_plan.py) → FanOut producers listed by the plan + (schema, lineage, stream-job, report, semantic, signal-gated orch/ELT/DQ) │ ▼ FanOut skeptics (adversarial workers / rubric scorers) @@ -342,15 +343,18 @@ Principle: **mirror or export → walk → specialize workers → promote → in ```text 1. Identify control plane + storage plane + compute plane + serve plane 2. Export or mirror artifacts into a local tree (or mount readonly) -3. Register SourceSystem(s) per plane entrypoint -4. Fan-out: +3. Plan (dekc_plan.py): ranked areas + signal-gated specialists +4. Register SourceSystem(s) per plane entrypoint +5. Fan-out only areas the plan listed: a. schema-scout → structural catalogs - b. stream-job-scout → landing producers + b. stream-job-scout + orch/ELT specialists → landing producers c. lineage-tracer → edges across jobs/SQL d. report-cataloger → serve plane -5. semantic-mapper → gold/curated → business objects -6. layer-auditor → score + gaps -7. synthesizer → index + packs + receipt + e. duckdb / notebook / dq scouts when markers exist +6. semantic-mapper → gold/curated → business objects +7. Fan-out skeptics → re-adversary-judge (threshold 0.75) +8. layer-auditor → score + gaps +9. synthesizer → index + packs + receipt ``` ### 4.2 Azure Fabric (typical setup) diff --git a/docs/user_guide/user-guide.md b/docs/user_guide/user-guide.md index a74074d..3c2f4d5 100644 --- a/docs/user_guide/user-guide.md +++ b/docs/user_guide/user-guide.md @@ -10,7 +10,7 @@ truth_state: current **Data Engineering Knowledge Capture** turns data platforms into a durable, Git-native [OKF](https://github.com/SpillwaveSolutions/okf-plugin) knowledge graph, with multi-agent walk loops designed using [AGER](https://github.com/SpillwaveSolutions/okf-agent-graph) (OKF Agent Graph Engineering Runtime). -Plugin release **0.5.0**. Storage format is OKF **0.2**. Agent loops follow AGER **0.3** roles (orchestrator / worker / judge / synthesizer) even when you run DEKC skills without a separate AGER bundle. Search and pack use a disposable [retrieval ladder](../designs/retrieval-ladder.md) (SQLite index → ripgrep → scan); Git + Markdown stays source of truth. +Plugin release **0.5.3**. Storage format is OKF **0.2**. Agent loops follow AGER **0.3** roles (orchestrator / worker / judge / synthesizer) even when you run DEKC skills without a separate AGER bundle. Search and pack use a disposable [retrieval ladder](../designs/retrieval-ladder.md) (SQLite index → ripgrep → scan); Git + Markdown stays source of truth. Existing second brains: [noun-ownership migration](./noun-ownership-migration.md) (`Workflow` jobs → `IngestionJob`; diagrams stay SAC). @@ -134,7 +134,7 @@ Each intent returns: design checklist, ranked schema-typed concepts, progressive For Q&A, spawn **data-retriever** (`/dekc-retrieve`) instead of dumping a brain pack into the parent. -Skills: `dekc-retrieve`, `dekc-second-brain`, `dekc-design-report`, `dekc-land-data`. +Skills: `dekc-plan`, `dekc-retrieve`, `dekc-second-brain`, `dekc-design-report`, `dekc-land-data`. Patterns: `patterns/design-report-from-gold.md`, `patterns/land-stream-to-bronze.md`. @@ -158,16 +158,19 @@ cp "${CLAUDE_PLUGIN_ROOT}/.dekc/config.example.yml" .dekc/config.yml # edit knowledge_root, walk.max_files, promote.default_layer ``` -### 2. Walk a filesystem export of your platform +### 2. Plan, then walk a filesystem export of your platform -Point the walker at SQL models, dbt, notebooks export, or a local mirror of lake paths: +Point the planner at SQL models, dbt, notebooks export, or a local mirror of lake paths. Review the ranked areas, then walk (or `--from-plan --area`): ```bash +python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_orchestrate.py" \ + --repo . --system "Retail Lake" --scan-root ./lake-mirror --plan-only +# review knowledge/.dekc/re-plan.md — spawn only listed specialists python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_walk.py" ./lake-mirror \ --repo . --bundle knowledge --source-name retail-lake ``` -Or `/dekc-walk` with path arguments. The walker: +Or `/dekc-plan` then `/dekc-walk` with path arguments. The walker: 1. Registers a **SourceSystem** 2. Discovers `*.sql`, `*.dax`, parquet directories, medallion folder names @@ -235,6 +238,7 @@ If `okf-graph-eng` is available, prefer its pack/impact for the same paths. | Skill / command | Purpose | |-----------------|---------| | `dekc-init` | Scaffold knowledge catalogs + medallion layers | +| `dekc-plan` | Breadth-first RE plan + checklists (pause before fan-out) | | `dekc-walk` | Filesystem / SQL root discovery | | `dekc-capture-table` | Manual table + columns | | `dekc-capture-query` | SQL or DAX query | @@ -254,11 +258,13 @@ If `okf-graph-eng` is available, prefer its pack/impact for the same paths. | Agent | Use when | |-------|----------| | **data-retriever** | Query-time Q&A — spawn via `/dekc-retrieve`; parent keeps a card only | -| **data-lake-walker** | Full reverse-engineer loop (orchestrator) | +| **data-lake-walker** | Full reverse-engineer loop (orchestrator): plan → fan-out → judge | | **schema-scout** | Only structure (schemas/tables/columns) | | **lineage-tracer** | Only SQL/job lineage and promotions | | **semantic-mapper** | Business objects, glossary, metrics | | **report-cataloger** | Dashboards, reports, DAX | +| **airflow-scout** / **glue-job-scout** / **fabric-pipeline-scout** | Signal-gated orchestration (plan must list them) | +| **dbt-elt-scout** / **duckdb-scout** / **notebook-scout** / **dq-scout** | Signal-gated ELT / DQ | | **layer-auditor** | Judge medallion health and orphans | In Claude/Grok, ask for the agent by name or describe the walk (“walk this lake and promote gold tables”). In AGER terms, the walker is an **OrchestratorAgent**; scouts are **WorkerAgents**; the auditor is a **JudgeAgent**. diff --git a/hosts/deep-agents/SKILL.md b/hosts/deep-agents/SKILL.md index 0ef41d1..aa1e3d4 100644 --- a/hosts/deep-agents/SKILL.md +++ b/hosts/deep-agents/SKILL.md @@ -21,3 +21,5 @@ export SECOND_BRAIN_ROOT="${SECOND_BRAIN_ROOT:-knowledge}" ``` Open an isolation session before writing a shared institutional tree. + +Reverse-engineering: `dekc_plan.py` / `--plan-only`, then `--from-plan --area`. For Q&A, spawn `data-retriever`. diff --git a/hosts/grok-bot/SKILL.md b/hosts/grok-bot/SKILL.md index 1011d80..0152107 100644 --- a/hosts/grok-bot/SKILL.md +++ b/hosts/grok-bot/SKILL.md @@ -9,6 +9,6 @@ Read `docs/ONBOARDING.md` first, then follow `docs/GROK_BOT.md`. 1. Identity: `grok-bot/data-engineering-knowledge-capture` 2. Open an isolation session before knowledge writes (`scripts/brain_session.py open`) unless the human already pointed `SECOND_BRAIN_ROOT` at a session worktree. -3. For Q&A, spawn **data-retriever** (`/dekc-retrieve`); keep the card only. Pack 2 hops before writing owned types via this plugin's scripts. +3. Reverse-engineering: `dekc_plan.py` / `--plan-only`, then area fan-out (`--from-plan --area`). For Q&A, spawn **data-retriever** (`/dekc-retrieve`); keep the card only. Pack 2 hops before writing owned types via this plugin's scripts. 4. Close the session to PR. Report path + validation result. 5. Never document a private remote. Never write raw Markdown into the tree. diff --git a/marketplace.json b/marketplace.json index 8e90829..3121e03 100644 --- a/marketplace.json +++ b/marketplace.json @@ -11,7 +11,7 @@ "name": "data-engineering-knowledge-capture", "source": "./", "description": "Data lake second brain: schemas, lineage, medallion, semantic layer, glossary. Depends on PKC + OKF.", - "version": "0.5.2", + "version": "0.5.3", "author": { "name": "Rick Hightower" }, diff --git a/package.json b/package.json index 0e509cd..5f77f1a 100644 --- a/package.json +++ b/package.json @@ -13,7 +13,7 @@ "build:dev": "vite build --mode development", "preview": "vite preview --host 0.0.0.0 --port 8080", "typecheck": "tsc --noEmit", - "test": "python3 tests/test_dekc.py && python3 tests/test_retrieval_ladder.py && python3 tests/test_fabric_walk_fixes.py && python3 tests/test_isolation.py", + "test": "python3 tests/test_dekc.py && python3 tests/test_dekc_plan.py && python3 tests/test_retrieval_ladder.py && python3 tests/test_fabric_walk_fixes.py && python3 tests/test_isolation.py", "lint": "eslint .", "format": "prettier --write .", "test:js": "node --test 'scripts/**/*.test.mjs'", @@ -99,6 +99,6 @@ "vite": "^8.2.0" }, "description": "Data Engineering Knowledge Capture (DEKC) \u2014 Claude/Grok/Codex/OpenCode plugin + explorer", - "version": "0.5.2", + "version": "0.5.3", "license": "MIT" } diff --git a/plugin.json b/plugin.json index cd3c341..60559bf 100644 --- a/plugin.json +++ b/plugin.json @@ -1,7 +1,7 @@ { "$schema": "https://agent-plugins.org/schemas/1.0.0/plugin.schema.json", "name": "data-engineering-knowledge-capture", - "version": "0.5.2", + "version": "0.5.3", "description": "Data Engineering Knowledge Capture: schemas, lineage, medallion, semantic models, and glossary into a durable OKF knowledge graph. Multi-host bindings and write isolation.", "author": { "name": "Rick Hightower", diff --git a/public/data/catalog.json b/public/data/catalog.json index bbf6417..ac65b37 100644 --- a/public/data/catalog.json +++ b/public/data/catalog.json @@ -1,6 +1,6 @@ { "title": "Retail Lake Sample Knowledge", - "version": "0.5.2", + "version": "0.5.3", "depends_on": [ "project-knowledge-capture", "okf-graph-eng" diff --git a/scripts/dekc_orchestrate.py b/scripts/dekc_orchestrate.py new file mode 100644 index 0000000..d46a62f --- /dev/null +++ b/scripts/dekc_orchestrate.py @@ -0,0 +1,283 @@ +#!/usr/bin/env python3 +"""Top-level reverse-engineering orchestrator. + +Init + breadth-first plan, then optional pause, then domain-scoped walk/capture. +Agent hosts invoke this as the deterministic backbone of +reverse-engineering-orchestrator / data-lake-walker. + +Workflow: init-bundle → plan → (optional pause) → scoped walk → (agents enrich +and mark checklists) → existing grade / skeptics / judge. + +Query-time retrieve (data-retriever / dekc-retrieve) is a different path. +No live Glue/S3/Fabric control-plane calls. +""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path +from typing import Any + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +from dekc_common import append_log, ensure_bundle, resolve_author, resolve_knowledge_root # noqa: E402 +from dekc_plan import ( # noqa: E402 + checklist_summary, + load_plan, + mark_area_item_if_present, + plan_paths, + scan_domains_from_plan, + write_plan, +) +from dekc_validate import validate_bundle # noqa: E402 +from dekc_walk import walk_scoped # noqa: E402 + + +def _public_plan(plan: dict[str, Any]) -> dict[str, Any]: + return { + "system": plan.get("system"), + "system_slug": plan.get("system_slug"), + "roots": plan.get("roots"), + "exports": plan.get("exports"), + "ecosystems": plan.get("ecosystems"), + "focus_areas": [ + { + "id": a.get("id"), + "rank": a.get("rank"), + "title": a.get("title"), + "kind": a.get("kind") or "domain", + "parent": a.get("parent"), + "signal": a.get("signal"), + "hit_count": a.get("hit_count"), + "agent": a.get("agent"), + "scan_domains": a.get("scan_domains"), + "spawn": a.get("spawn", True), + "checklist": a.get("checklist"), + } + for a in (plan.get("focus_areas") or []) + ], + "specialists": plan.get("specialists") + or [ + { + "id": a.get("id"), + "kind": a.get("kind"), + "parent": a.get("parent"), + "agent": a.get("agent"), + "title": a.get("title"), + "signal": a.get("signal"), + } + for a in (plan.get("focus_areas") or []) + if a.get("kind") in ("orchestration", "elt-tool") + ], + "cicd_handoff": plan.get("cicd_handoff") or [], + "artifacts": plan.get("artifacts"), + "written": plan.get("written"), + "checklist": plan.get("checklist") or checklist_summary(plan), + } + + +def _resolve_from_plan(bundle: Path, from_plan: str | Path | None) -> Path: + if from_plan: + p = Path(from_plan) + if not p.is_absolute(): + if p.exists(): + return p.resolve() + cand = bundle / p + if cand.exists(): + return cand.resolve() + return p.resolve() + return p + return plan_paths(bundle)["json"] + + +def orchestrate( + host_repo: Path, + scan_roots: list[Path], + *, + system_name: str, + bundle_name: str | None, + author: str, + exports: list[Path] | None = None, + fabric_items: Path | None = None, + pbi_bindings: Path | None = None, + inventory: Path | None = None, + workspace: str = "", + workspace_id: str = "", + inventory_layer: str = "gold", + plan_only: bool = False, + from_plan: str | Path | None = None, + area: str | None = None, + dry_run: bool = False, +) -> dict[str, Any]: + bundle = resolve_knowledge_root(host_repo, bundle_name) + if not dry_run: + ensure_bundle(bundle, system_name) + else: + bundle.mkdir(parents=True, exist_ok=True) + phases_done = ["init-bundle"] + + export_paths = list(exports or []) + if fabric_items: + export_paths.append(fabric_items) + if pbi_bindings: + export_paths.append(pbi_bindings) + if inventory: + export_paths.append(inventory) + + if from_plan: + plan = load_plan(_resolve_from_plan(bundle, from_plan)) + if system_name == "Data platform" and plan.get("system"): + system_name = plan["system"] + else: + plan = write_plan(bundle, scan_roots, system_name=system_name, exports=export_paths) + phases_done.append("plan") + append_log( + bundle, + "RE plan: areas=" + + ",".join(a["id"] for a in (plan.get("focus_areas") or [])) + + f" checklist={checklist_summary(plan)}", + ) + + if plan_only: + validation = validate_bundle(bundle) + return { + "bundle": str(bundle), + "system": system_name, + "phases": phases_done, + "plan": _public_plan(plan), + "walk": None, + "validation": { + "ok": validation["ok"], + "errors": validation["errors"], + "warnings": validation["warnings"], + }, + } + + domains = scan_domains_from_plan(plan, area=area) + focus_match = next((a for a in (plan.get("focus_areas") or []) if a.get("id") == area), None) + enrichment_only = bool(area and focus_match is not None and not (focus_match.get("scan_domains") or [])) + lake_root = scan_roots[0] if scan_roots else None + walk_payload: dict[str, Any] | None + if enrichment_only: + walk_payload = {"created": [], "updated": [], "skipped": [], "errors": [], "domains": [], "counts": {}} + phases_done.append("enrichment") + else: + result = walk_scoped( + lake_root, + bundle, + domains=domains, + source_name=system_name, + fabric_items=fabric_items, + pbi_bindings=pbi_bindings, + inventory=inventory, + workspace=workspace, + workspace_id=workspace_id, + inventory_layer=inventory_layer, + dry_run=dry_run, + ) + walk_payload = result.to_dict() + walk_payload["domains"] = domains + phases_done.append("walk") + if area: + mark_area_item_if_present( + bundle, area=area, item="capture", status="done", note="orchestrate --from-plan" + ) + else: + for focus in plan.get("focus_areas") or []: + if "capture" in {i["id"] for i in (focus.get("checklist") or [])}: + mark_area_item_if_present( + bundle, + area=focus["id"], + item="capture", + status="done", + note="orchestrate scoped capture", + ) + try: + plan = load_plan(bundle) + except FileNotFoundError: + pass + validation = validate_bundle(bundle) + phases_done.append("validate") + append_log(bundle, f"Orchestrate complete: phases={','.join(phases_done)}") + return { + "bundle": str(bundle), + "system": system_name, + "phases": phases_done, + "plan": _public_plan(plan), + "walk": walk_payload, + "validation": { + "ok": validation["ok"], + "errors": validation["errors"], + "warnings": validation["warnings"], + }, + } + + +def main(argv: list[str] | None = None) -> int: + p = argparse.ArgumentParser(description="DEKC reverse-engineering orchestrator") + p.add_argument("--repo", default=".", help="Knowledge host repo") + p.add_argument("--bundle", default=None) + p.add_argument("--system", default="Data platform") + p.add_argument("--scan-root", action="append", default=[], help="Lake / SQL / job root(s)") + p.add_argument("--export", action="append", default=[], help="Optional export JSON paths") + p.add_argument("--fabric-items", default=None) + p.add_argument("--pbi-bindings", default=None) + p.add_argument("--inventory", default=None) + p.add_argument("--workspace", default="") + p.add_argument("--workspace-id", default="") + p.add_argument("--inventory-layer", default="gold") + p.add_argument("--plan-only", action="store_true", help="Init + breadth-first plan, then stop") + p.add_argument("--from-plan", default=None, help="Existing .dekc/re-plan.json") + p.add_argument("--area", default=None, help="With --from-plan: one focus area") + p.add_argument("--dry-run", action="store_true") + p.add_argument("--json", action="store_true") + p.add_argument("--author", default="") + args = p.parse_args(argv) + author = "" + if not args.dry_run: + author = resolve_author(args.author) + host = Path(args.repo).resolve() + roots = [Path(r).resolve() for r in (args.scan_root or ([str(host)] if not args.from_plan else []))] + result = orchestrate( + host, + roots, + system_name=args.system, + bundle_name=args.bundle, + author=author, + exports=[Path(e).resolve() for e in (args.export or [])], + fabric_items=Path(args.fabric_items).resolve() if args.fabric_items else None, + pbi_bindings=Path(args.pbi_bindings).resolve() if args.pbi_bindings else None, + inventory=Path(args.inventory).resolve() if args.inventory else None, + workspace=args.workspace, + workspace_id=args.workspace_id, + inventory_layer=args.inventory_layer, + plan_only=args.plan_only, + from_plan=args.from_plan, + area=args.area, + dry_run=args.dry_run, + ) + if args.json: + print(json.dumps(result, indent=2, default=str)) + else: + print("Data Engineering Knowledge Capture orchestrate") + print(f" bundle: {result['bundle']}") + print(f" system: {result['system']}") + print(f" phases: {', '.join(result['phases'])}") + print(f" valid: {result['validation']['ok']}") + plan = result.get("plan") or {} + areas = ", ".join( + f"{a.get('rank')}:{a.get('id')}({a.get('agent')})" for a in (plan.get("focus_areas") or []) + ) + if areas: + print(f" plan: {areas}") + if plan.get("checklist"): + print(f" checks: {plan['checklist']}") + written = plan.get("written") or {} + if written.get("md"): + print(f" plan md: {written['md']}") + return 0 if result["validation"]["ok"] else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/dekc_plan.py b/scripts/dekc_plan.py new file mode 100644 index 0000000..0a5bb9d --- /dev/null +++ b/scripts/dekc_plan.py @@ -0,0 +1,1336 @@ +#!/usr/bin/env python3 +"""Breadth-first reverse-engineering plan (presence and counts only). + +Maps scan roots plus optional control-plane export files, ranks focus areas, +and emits a Markdown checklist plus JSON the orchestrator and specialists consume. + +This is not a full walk: no live Glue/S3/Fabric calls, no GE runtime. +Query-time retrieve (data-retriever / dekc-retrieve) is a different path. + +DEKC owns data orchestration + ELT/ETL. SAC owns CI/CD. Actions that only +trigger Glue/dbt are recorded as a cross-link note, not a Pipeline noun. +""" + +from __future__ import annotations + +import argparse +import json +import sys +from collections import deque +from pathlib import Path +from typing import Any + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +from dekc_common import slugify # noqa: E402 + +PLAN_VERSION = "1" +DEKC_DIRNAME = ".dekc" +PLAN_JSON_NAME = "re-plan.json" +PLAN_MD_NAME = "re-plan.md" +PLAN_PROGRESS_NAME = "re-plan-progress.json" + +DOT_WALK = frozenset({".github", ".gitlab", ".circleci"}) +PLAN_IGNORE = frozenset( + { + "node_modules", + "__pycache__", + ".venv", + "venv", + "dist", + "build", + "target", + ".tox", + ".mypy_cache", + ".pytest_cache", + ".idea", + ".vscode", + ".terraform", + ".git", + ".grok", + ".output", + ".tanstack", + ".nitro", + ".vinxi", + } +) + +MAX_DEPTH = 5 +MAX_FILES = 2500 +PEEK_BYTES = 2048 +MAX_HITS = 24 + +MAJOR_DIR_NAMES = { + "bronze", + "silver", + "gold", + "raw", + "staging", + "curated", + "marts", + "models", + "dags", + "airflow", + "glue", + "adf", + "pipelines", + "notebooks", + "sql", + "warehouse", + "lake", + "lakehouse", + "dbt", + "spark", + "emr", + "quality", + "gx", + "great_expectations", + "soda", + "storage", + "catalog", + "k8s", + "cron", + "composer", + "stepfunctions", + "analytics", +} + +LAKE_DIR_HINTS = { + "bronze", + "silver", + "gold", + "raw", + "staging", + "curated", + "marts", + "lake", + "lakehouse", + "warehouse", +} + +SQL_SUFFIXES = {".sql"} +DAX_SUFFIXES = {".dax"} +YAML_SUFFIXES = {".yml", ".yaml"} +JSON_SUFFIXES = {".json"} + + +def _checklist(pairs: list[tuple[str, str]]) -> list[dict[str, str]]: + return [{"id": i, "text": t, "status": "pending", "note": ""} for i, t in pairs] + + +AREA_SPECS: dict[str, dict[str, Any]] = { + "lake": { + "title": "Lake / SQL / parquet / Delta", + "kind": "domain", + "agent": "schema-scout", + "scan_domains": ["lake"], + "weight": 3, + "checklist": _checklist( + [ + ("inventory", "Inventory SQL trees, parquet datasets, and Delta _delta_log markers"), + ("layers", "Map medallion folders (bronze / silver / gold / raw)"), + ("capture", "Capture tables/views/queries via filesystem walk (script-owned write)"), + ("stubs", "Record referenced tables as stubs only when SQL evidence exists"), + ("enrich", "Enrich schema/column contracts (agent judgment; no invented columns)"), + ] + ), + }, + "elt": { + "title": "ELT / ETL (dbt, Spark/EMR, warehouse SQL)", + "kind": "domain", + "agent": "stream-job-scout", + "scan_domains": ["elt"], + "weight": 3, + "checklist": _checklist( + [ + ("inventory", "Inventory dbt projects, Spark/EMR job defs, and warehouse SQL trees"), + ("capture", "Capture IngestionJob / Transformation markers (script-owned write)"), + ("models", "Note dbt models vs ad-hoc warehouse SQL"), + ("link", "Join jobs to landing tables only when evidenced"), + ("enrich", "Enrich purpose and cadence; do not invent lineage"), + ] + ), + }, + "orchestration": { + "title": "Data orchestration (Airflow / ADF / Glue / Fabric / cron)", + "kind": "domain", + "agent": "stream-job-scout", + "scan_domains": ["orchestration"], + "weight": 4, + "checklist": _checklist( + [ + ("inventory", "Inventory Airflow DAGs, ADF, Glue scripts, Step Functions, Fabric pipelines, cron loaders"), + ("capture", "Capture IngestionJob concepts for evidenced orchestrators (script-owned write)"), + ("schedule", "Note schedules / DAG ids / state machines when present in files"), + ("link", "Join jobs to bronze/silver landings only with evidence"), + ("enrich", "Enrich ownership; SAC owns CI/CD — do not capture Actions as Pipeline"), + ] + ), + }, + "catalogs": { + "title": "Catalogs / StorageLocation", + "kind": "domain", + "agent": "schema-scout", + "scan_domains": ["catalogs"], + "weight": 2, + "checklist": _checklist( + [ + ("inventory", "Inventory catalog JSON / INFORMATION_SCHEMA exports and storage path conventions"), + ("capture", "Capture DataCatalog / StorageLocation (script-owned write)"), + ("link", "Join storage to layers when path prefixes evidence a zone"), + ("enrich", "Do not call live Glue/S3 APIs — exports and mirrors only"), + ] + ), + }, + "bi": { + "title": "Reports / DAX / Power BI / Fabric serve", + "kind": "domain", + "agent": "report-cataloger", + "scan_domains": ["bi"], + "weight": 3, + "checklist": _checklist( + [ + ("inventory", "Inventory DAX, Power BI bindings, and Fabric Report / SemanticModel exports"), + ("capture", "Capture Report / Dashboard / SemanticModel (script-owned write)"), + ("bind", "Join reports to datasets only when binding JSON evidences it"), + ("enrich", "Fabric Report ≠ DEKC Dashboard unless the export says Dashboard"), + ] + ), + }, + "lineage": { + "title": "SQL / job lineage", + "kind": "domain", + "agent": "lineage-tracer", + "scan_domains": [], + "weight": 2, + "checklist": _checklist( + [ + ("sql", "Extract FROM/JOIN edges from walked SQL (no invented edges)"), + ("jobs", "Materialize job → table landings only with evidence"), + ("promote", "Record bronze→silver→gold promotions when the same basename appears"), + ("enrich", "Retract unproven edges rather than inventing them for grade"), + ] + ), + }, + "semantic": { + "title": "Business objects / gold semantics", + "kind": "domain", + "agent": "semantic-mapper", + "scan_domains": [], + "weight": 2, + "checklist": _checklist( + [ + ("gold", "Identify gold / mart tables and curated semantic models"), + ("promote", "Promote gold → BusinessObject + glossary when meaning is evidenced"), + ("metrics", "Capture metrics only with DAX/SQL definitions"), + ("enrich", "Do not invent business names for bronze stubs"), + ] + ), + }, + "dq": { + "title": "Data quality markers (GE / Soda / dbt tests)", + "kind": "domain", + "agent": "dq-scout", + "scan_domains": ["dq"], + "weight": 3, + "checklist": _checklist( + [ + ("inventory", "Inventory Great Expectations suites, Soda YAML, and dbt tests"), + ("capture", "Capture DQRule nouns from filenames / suite ids (cheap; no GE runtime)"), + ("link", "Join rules to tables only when the suite names a table"), + ("enrich", "Do not fake expectation results or Soda scan output"), + ] + ), + }, +} + +SPECIALIST_SPECS: dict[str, dict[str, Any]] = { + "orch-airflow": { + "title": "Airflow / DAG folders", + "kind": "orchestration", + "parent": "orchestration", + "agent": "airflow-scout", + "scan_domains": [], + "weight": 4, + "checklist": _checklist( + [ + ("inventory", "Inventory dags/ and DAG( / airflow imports"), + ("dag-ids", "Record dag_id values from source when present"), + ("enrich", "Enrich IngestionJob purpose (do not re-walk the whole lake)"), + ("link", "Join DAGs to landing tables only with SQL/task evidence"), + ] + ), + }, + "orch-glue": { + "title": "Glue job scripts", + "kind": "orchestration", + "parent": "orchestration", + "agent": "glue-job-scout", + "scan_domains": [], + "weight": 4, + "checklist": _checklist( + [ + ("inventory", "Inventory Glue job scripts (awsglue / GlueJob path conventions)"), + ("enrich", "Enrich IngestionJob purpose from comments / job names"), + ("link", "Join jobs to S3/StorageLocation only when the script names a path"), + ("boundary", "No live Glue API — export/mirror only"), + ] + ), + }, + "orch-fabric": { + "title": "Fabric pipelines / notebooks (export)", + "kind": "orchestration", + "parent": "orchestration", + "agent": "fabric-pipeline-scout", + "scan_domains": [], + "weight": 4, + "checklist": _checklist( + [ + ("inventory", "Inventory Fabric pipeline / notebook items from export JSON"), + ("enrich", "Enrich IngestionJob (fabric-pipeline vs fabric-notebook)"), + ("serve", "Leave Report / SemanticModel to report-cataloger"), + ("boundary", "dekc_walk does not call Fabric REST"), + ] + ), + }, + "orch-adf": { + "title": "Azure Data Factory", + "kind": "orchestration", + "parent": "orchestration", + "agent": "adf-scout", + "scan_domains": [], + "weight": 3, + "checklist": _checklist( + [ + ("inventory", "Inventory ADF pipeline ARM / factory JSON"), + ("enrich", "Enrich IngestionJob orchestrator=adf"), + ] + ), + }, + "orch-stepfunctions": { + "title": "Step Functions ASL", + "kind": "orchestration", + "parent": "orchestration", + "agent": "stepfunctions-scout", + "scan_domains": [], + "weight": 3, + "checklist": _checklist( + [ + ("inventory", "Inventory ASL state machines (StartAt + States)"), + ("enrich", "Enrich IngestionJob orchestrator=stepfunctions"), + ] + ), + }, + "orch-composer": { + "title": "Cloud Composer", + "kind": "orchestration", + "parent": "orchestration", + "agent": "composer-scout", + "scan_domains": [], + "weight": 3, + "checklist": _checklist( + [ + ("inventory", "Inventory Composer markers (composer.googleapis.com / composer/)"), + ("airflow", "Composer still runs Airflow — do not drop airflow-scout if DAGs exist"), + ("enrich", "Enrich IngestionJob orchestrator=composer"), + ] + ), + }, + "orch-cron": { + "title": "Cron / K8s CronJob loaders", + "kind": "orchestration", + "parent": "orchestration", + "agent": "cron-loader-scout", + "scan_domains": [], + "weight": 3, + "checklist": _checklist( + [ + ("inventory", "Inventory crontab and kind: CronJob loaders (dbt / spark-submit / python)"), + ("enrich", "Enrich IngestionJob orchestrator=cron or k8s-cronjob"), + ("skip-ci", "Scheduled GitHub Actions stay on SAC unless they only trigger a data job"), + ] + ), + }, + "elt-dbt": { + "title": "dbt ELT", + "kind": "elt-tool", + "parent": "elt", + "agent": "dbt-elt-scout", + "scan_domains": [], + "weight": 4, + "checklist": _checklist( + [ + ("project", "Map dbt_project.yml, models/, and sources"), + ("tests", "Note dbt tests (hand to dq-scout; do not fake run results)"), + ("enrich", "Enrich IngestionJob / Transformation purpose (walk already captured model SQL)"), + ] + ), + }, + "duckdb": { + "title": "DuckDB", + "kind": "elt-tool", + "parent": "lake", + "agent": "duckdb-scout", + "scan_domains": ["duckdb"], + "weight": 3, + "checklist": _checklist( + [ + ("inventory", "Inventory *.duckdb / *.ddb and duckdb SQL"), + ("capture", "Capture SourceSystem kind=duckdb (script-owned write)"), + ("enrich", "Do not open the binary as a live catalog unless a SQL export exists"), + ] + ), + }, + "notebooks": { + "title": "Notebooks (.ipynb / Fabric notebooks)", + "kind": "elt-tool", + "parent": "elt", + "agent": "notebook-scout", + "scan_domains": ["notebooks"], + "weight": 3, + "checklist": _checklist( + [ + ("inventory", "Inventory .ipynb and Fabric notebook exports"), + ("sql", "Extract SQL cells (%%sql / spark.sql / SELECT) — cheap capture"), + ("enrich", "Do not invent lineage from incomplete notebook cells"), + ] + ), + }, +} + + +def plan_paths(bundle: Path) -> dict[str, Path]: + d = bundle / DEKC_DIRNAME + return { + "dir": d, + "json": d / PLAN_JSON_NAME, + "md": d / PLAN_MD_NAME, + "progress": d / PLAN_PROGRESS_NAME, + } + + +def resolve_plan_files(plan_or_bundle: Path) -> dict[str, Path]: + p = Path(plan_or_bundle) + if p.is_file() and p.suffix == ".json": + d = p.parent + return {"dir": d, "json": p, "md": d / PLAN_MD_NAME, "progress": d / PLAN_PROGRESS_NAME} + if p.is_dir() and (p / PLAN_JSON_NAME).is_file(): + return { + "dir": p, + "json": p / PLAN_JSON_NAME, + "md": p / PLAN_MD_NAME, + "progress": p / PLAN_PROGRESS_NAME, + } + return plan_paths(p) + + +def _rel(root: Path, path: Path) -> str: + try: + return str(path.relative_to(root)).replace("\\", "/") + except ValueError: + return str(path) + + +def _should_skip(name: str) -> bool: + if name in DOT_WALK: + return False + if name in PLAN_IGNORE: + return True + if name.startswith(".git"): + return True + if name.startswith(".") and name not in DOT_WALK: + return True + return False + + +def bfs_walk(root: Path, *, max_depth: int = MAX_DEPTH, max_files: int = MAX_FILES) -> list[Path]: + """Breadth-first file listing. Cheap presence scan, not a full lake dump.""" + root = root.resolve() + out: list[Path] = [] + q: deque[tuple[Path, int]] = deque([(root, 0)]) + while q and len(out) < max_files: + cur, depth = q.popleft() + if depth > max_depth: + continue + try: + entries = sorted(cur.iterdir(), key=lambda p: (not p.is_dir(), p.name.lower())) + except (OSError, PermissionError): + continue + child_dirs: list[Path] = [] + for p in entries: + if _should_skip(p.name): + continue + if p.is_dir(): + child_dirs.append(p) + elif p.is_file(): + out.append(p) + if len(out) >= max_files: + break + if depth < max_depth: + for d in child_dirs: + q.append((d, depth + 1)) + return out + + +def _peek(path: Path, limit: int = PEEK_BYTES) -> str: + try: + if path.stat().st_size > 1_000_000: + return "" + with path.open("r", encoding="utf-8", errors="replace") as fh: + return fh.read(limit) + except OSError: + return "" + + +def _name_has_any(name: str, needles: tuple[str, ...]) -> bool: + low = name.lower() + return any(n in low for n in needles) + + +def _hit(kind: str, rel: str, extra: dict[str, Any] | None = None) -> dict[str, Any]: + row = {"kind": kind, "path": rel} + if extra: + row.update(extra) + return row + + +def _trim_hits(groups: dict[str, list[dict[str, Any]]]) -> dict[str, list[dict[str, Any]]]: + trimmed: dict[str, list[dict[str, Any]]] = {} + for key, rows in groups.items(): + seen: set[tuple[str, str]] = set() + uniq: list[dict[str, Any]] = [] + for row in rows: + k = (row.get("kind", ""), row.get("path", "")) + if k in seen: + continue + seen.add(k) + uniq.append(row) + trimmed[key] = uniq[:MAX_HITS] + return trimmed + + +def _looks_like_cicd_handoff(text: str) -> bool: + low = text.lower() + return any( + n in low + for n in ( + "glue", + "dbt ", + "dbt-", + "airflow", + "awsglue", + "databricks", + "spark-submit", + "fabric", + "datafactory", + ) + ) + + +def inspect_export(path: Path) -> dict[str, list[dict[str, Any]]]: + """Presence-only read of a control-plane JSON/YAML export (no live APIs).""" + hits: dict[str, list[dict[str, Any]]] = {k: [] for k in list(AREA_SPECS) + list(SPECIALIST_SPECS)} + path = Path(path) + if not path.is_file(): + return hits + rel = path.name + text = _peek(path, 8000) + low = text.lower() + suffix = path.suffix.lower() + if suffix not in {".json", ".yml", ".yaml"}: + return hits + + fabric_needles = ( + "lakehouse", + "datapipeline", + "semanticmodel", + "sqlendpoint", + "eventstream", + '"notebook"', + ) + if any(n in low for n in fabric_needles) and ("displayname" in low or '"type"' in low): + hits["orchestration"].append(_hit("fabric-export", rel)) + hits["orch-fabric"].append(_hit("fabric-export", rel)) + if any(n in low for n in ("report", "semanticmodel", "dashboard", "dataset")): + hits["bi"].append(_hit("fabric-serve", rel)) + if "notebook" in low: + hits["notebooks"].append(_hit("fabric-notebook", rel)) + + if any(n in text for n in ("TABLE_NAME", "table_name", "TABLE_SCHEMA", "information_schema")): + hits["catalogs"].append(_hit("inventory-json", rel)) + elif '"tables"' in low and ("schema" in low or "layer" in low): + hits["catalogs"].append(_hit("inventory-json", rel)) + + if "datasetid" in low or "dataset_id" in low or "datasources_status" in low: + hits["bi"].append(_hit("pbi-bindings", rel)) + + if "awstemplateformatversion" in low and "datafactory" in low: + hits["orchestration"].append(_hit("adf-export", rel)) + hits["orch-adf"].append(_hit("adf-export", rel)) + + return hits + + +def inspect_root(root: Path) -> dict[str, Any]: + root = root.resolve() + files = bfs_walk(root) + top_level: list[str] = [] + try: + for p in sorted(root.iterdir(), key=lambda x: x.name.lower()): + if p.name in PLAN_IGNORE or p.name == ".git": + continue + if p.name.startswith(".") and p.name not in DOT_WALK: + continue + top_level.append(p.name + ("/" if p.is_dir() else "")) + except OSError: + pass + + major_dirs: list[str] = [] + seen_major: set[str] = set() + hits: dict[str, list[dict[str, Any]]] = {k: [] for k in AREA_SPECS} + specialist_hits: dict[str, list[dict[str, Any]]] = {k: [] for k in SPECIALIST_SPECS} + ecosystems: set[str] = set() + cicd_handoff: list[dict[str, Any]] = [] + sql_count = 0 + parquet_count = 0 + delta_count = 0 + + for f in files: + rel = _rel(root, f) + parts = Path(rel).parts + name = f.name + suffix = f.suffix.lower() + low_name = name.lower() + low_rel = rel.lower() + + for part in parts[:-1]: + if part in MAJOR_DIR_NAMES and part not in seen_major: + seen_major.add(part) + major_dirs.append(part) + + if suffix in SQL_SUFFIXES: + sql_count += 1 + hits["lake"].append(_hit("sql", rel)) + ecosystems.add("sql") + peek = _peek(f) + if "duckdb" in peek.lower() or "duckdb" in low_rel: + specialist_hits["duckdb"].append(_hit("duckdb-sql", rel)) + ecosystems.add("duckdb") + if any(p in {"models", "dbt"} for p in parts) or "ref(" in peek: + hits["elt"].append(_hit("dbt-sql", rel)) + specialist_hits["elt-dbt"].append(_hit("dbt-model", rel)) + ecosystems.add("dbt") + elif any(p in {"warehouse", "marts", "gold", "silver"} for p in parts): + hits["elt"].append(_hit("warehouse-sql", rel)) + + if suffix in DAX_SUFFIXES or low_name.endswith(".dax.cs"): + hits["bi"].append(_hit("dax", rel)) + ecosystems.add("dax") + + if suffix == ".parquet" or low_name.endswith(".parquet"): + parquet_count += 1 + hits["lake"].append(_hit("parquet", rel)) + ecosystems.add("parquet") + + if name == "_delta_log" or "_delta_log" in parts: + delta_count += 1 + hits["lake"].append(_hit("delta", rel)) + ecosystems.add("delta") + if low_name == "000.json" and "_delta_log" in parts: + hits["lake"].append(_hit("delta-log", rel)) + ecosystems.add("delta") + + if suffix in {".duckdb", ".ddb"} or low_name.endswith(".duckdb"): + specialist_hits["duckdb"].append(_hit("duckdb-file", rel)) + hits["lake"].append(_hit("duckdb-file", rel)) + ecosystems.add("duckdb") + + if suffix == ".ipynb": + specialist_hits["notebooks"].append(_hit("ipynb", rel)) + hits["elt"].append(_hit("notebook", rel)) + ecosystems.add("notebook") + + if low_name in {"dbt_project.yml", "dbt_project.yaml", "packages.yml", "dependencies.yml"}: + hits["elt"].append(_hit("dbt-project", rel)) + specialist_hits["elt-dbt"].append(_hit("dbt-project", rel)) + ecosystems.add("dbt") + if low_name in {"schema.yml", "schema.yaml", "_schema.yml"} and any( + p in {"models", "dbt"} for p in parts + ): + peek = _peek(f) + specialist_hits["elt-dbt"].append(_hit("dbt-schema", rel)) + if "tests:" in peek or "data_tests:" in peek: + hits["dq"].append(_hit("dbt-tests", rel)) + + spark_name = _name_has_any(low_rel, ("spark", "emr", "glue")) + if suffix in {".py", ".scala", ".json"} and spark_name: + peek = _peek(f) if suffix != ".json" else _peek(f) + if any( + n in peek + for n in ("SparkSession", "spark-submit", "awsglue", "EmrJob", "glue.Job") + ) or any(p in {"spark", "emr"} for p in parts): + hits["elt"].append(_hit("spark-job", rel)) + ecosystems.add("spark") + + if "dags" in parts or low_name.startswith("dag_") or "airflow" in parts: + peek = _peek(f) if suffix in {".py", ".yml", ".yaml"} else "" + if suffix == ".py" or "airflow" in peek.lower() or "dags" in parts: + specialist_hits["orch-airflow"].append(_hit("airflow", rel)) + hits["orchestration"].append(_hit("airflow", rel)) + ecosystems.add("airflow") + elif suffix == ".py": + peek = _peek(f) + if "from airflow" in peek or "import airflow" in peek or "DAG(" in peek: + specialist_hits["orch-airflow"].append(_hit("airflow", rel)) + hits["orchestration"].append(_hit("airflow", rel)) + ecosystems.add("airflow") + + if suffix == ".py" and ( + "awsglue" in _peek(f) or "glue" in parts or low_name.startswith("glue_") + ): + peek = _peek(f) + if "awsglue" in peek or "GlueContext" in peek or "glue" in parts: + specialist_hits["orch-glue"].append(_hit("glue", rel)) + hits["orchestration"].append(_hit("glue", rel)) + ecosystems.add("glue") + + if suffix in JSON_SUFFIXES | YAML_SUFFIXES: + peek = _peek(f) + if '"StartAt"' in peek and '"States"' in peek: + specialist_hits["orch-stepfunctions"].append(_hit("stepfunctions", rel)) + hits["orchestration"].append(_hit("stepfunctions", rel)) + ecosystems.add("stepfunctions") + if "Microsoft.DataFactory" in peek or "datafactory" in low_rel: + specialist_hits["orch-adf"].append(_hit("adf", rel)) + hits["orchestration"].append(_hit("adf", rel)) + ecosystems.add("adf") + if any(n in peek.lower() for n in ("datapipeline", "fabric")) and ( + "pipeline" in low_name or "fabric" in low_rel + ): + specialist_hits["orch-fabric"].append(_hit("fabric-file", rel)) + hits["orchestration"].append(_hit("fabric-file", rel)) + ecosystems.add("fabric") + if "composer.googleapis.com" in peek or "composer" in parts: + specialist_hits["orch-composer"].append(_hit("composer", rel)) + hits["orchestration"].append(_hit("composer", rel)) + ecosystems.add("composer") + if "kind: CronJob" in peek or "kind: CronJob" in _peek(f): + cmdish = peek.lower() + if any(n in cmdish for n in ("dbt", "spark-submit", "python", "glue", "airflow")): + specialist_hits["orch-cron"].append(_hit("k8s-cronjob", rel)) + hits["orchestration"].append(_hit("k8s-cronjob", rel)) + ecosystems.add("cron") + + if low_name in {"crontab", "cron"} or "cron.d" in parts or low_name.endswith(".cron"): + peek = _peek(f) + if any(n in peek.lower() for n in ("dbt", "spark-submit", "python", "glue")): + specialist_hits["orch-cron"].append(_hit("cron", rel)) + hits["orchestration"].append(_hit("cron", rel)) + ecosystems.add("cron") + + if low_name in {"great_expectations.yml", "great_expectations.yaml"} or "great_expectations" in parts or "gx" in parts: + if suffix in YAML_SUFFIXES | JSON_SUFFIXES or low_name.endswith(".json"): + hits["dq"].append(_hit("great-expectations", rel)) + ecosystems.add("great-expectations") + if low_name.startswith("soda") or "soda" in parts: + if suffix in YAML_SUFFIXES: + hits["dq"].append(_hit("soda", rel)) + ecosystems.add("soda") + if "expectations" in parts and suffix == ".json": + hits["dq"].append(_hit("ge-suite", rel)) + ecosystems.add("great-expectations") + + if any(n in low_rel for n in ("s3://", "abfss://", "wasbs://")) or any( + p in {"storage", "s3", "adls", "warehouse"} for p in parts + ): + if suffix in {".sql", ".py", ".json", ".yml", ".yaml", ".md"} or name.endswith(".parquet"): + hits["catalogs"].append(_hit("storage-path", rel)) + if any(p in {"catalog", "catalogs"} for p in parts) or low_name in { + "catalog.json", + "inventory.json", + "information_schema.json", + }: + hits["catalogs"].append(_hit("catalog", rel)) + + if ".github" in parts and "workflows" in parts and suffix in YAML_SUFFIXES: + peek = _peek(f) + if _looks_like_cicd_handoff(peek): + cicd_handoff.append(_hit("github-actions", rel, {"owner": "sac"})) + + if any(d in LAKE_DIR_HINTS for d in seen_major): + for d in sorted(seen_major & LAKE_DIR_HINTS): + hits["lake"].append(_hit("layer-dir", d + "/")) + + if sql_count or parquet_count or delta_count or hits["elt"] or hits["orchestration"]: + hits["lineage"].append( + _hit( + "sql-or-jobs", + f"{sql_count} sql / {len(hits['orchestration'])} orch", + {"count": sql_count + parquet_count + len(hits["orchestration"])}, + ) + ) + + if any(d in {"gold", "marts", "semantic"} for d in seen_major) or hits["bi"]: + hits["semantic"].append(_hit("gold-or-bi", "gold/marts or BI signals")) + + if sql_count: + hits["lake"].append(_hit("source-files", f"{sql_count} sql", {"count": sql_count})) + + if "bronze" in seen_major or "gold" in seen_major or parquet_count or delta_count: + layout = "lake" + elif "dbt" in ecosystems and not hits["orchestration"]: + layout = "dbt-project" + elif hits["orchestration"] and not hits["lake"]: + layout = "orchestration" + elif sql_count and not parquet_count: + layout = "warehouse" + else: + layout = "mixed" + + return { + "path": str(root), + "name": root.name, + "top_level": top_level, + "major_dirs": major_dirs, + "ecosystems": sorted(ecosystems), + "layout": layout, + "hits": _trim_hits(hits), + "specialist_hits": _trim_hits(specialist_hits), + "cicd_handoff": cicd_handoff[:MAX_HITS], + "file_count": len(files), + "sql_count": sql_count, + "parquet_count": parquet_count, + "delta_count": delta_count, + } + + +def _signal_for(area_id: str, spec: dict[str, Any], rows: list[dict[str, Any]]) -> int: + weight = int(spec["weight"]) + n = 0 + for row in rows: + if row.get("kind") in {"source-files", "sql-or-jobs"}: + n += min(int(row.get("count") or 0), 8) + else: + n += 1 + return n * weight + + +def _combine_hits(inspected: list[dict[str, Any]], key: str) -> dict[str, list[dict[str, Any]]]: + combined: dict[str, list[dict[str, Any]]] = {} + for info in inspected: + for area_id, rows in (info.get(key) or {}).items(): + bucket = combined.setdefault(area_id, []) + prefix = info["name"] + for row in rows: + item = dict(row) + item["root"] = prefix + bucket.append(item) + return combined + + +def _area_entry(area_id: str, spec: dict[str, Any], rows: list[dict[str, Any]]) -> dict[str, Any] | None: + signal = _signal_for(area_id, spec, rows) + if signal <= 0: + return None + return { + "id": area_id, + "title": spec["title"], + "kind": spec.get("kind") or "domain", + "parent": spec.get("parent"), + "signal": signal, + "hit_count": len(rows), + "agent": spec["agent"], + "scan_domains": list(spec.get("scan_domains") or []), + "hits": rows[:MAX_HITS], + "checklist": [dict(x) for x in (spec.get("checklist") or [])], + "spawn": True, + } + + +def build_plan( + roots: list[Path], + *, + system_name: str, + exports: list[Path] | None = None, +) -> dict[str, Any]: + inspected = [inspect_root(Path(r)) for r in roots] + combined = _combine_hits(inspected, "hits") + combined_spec = _combine_hits(inspected, "specialist_hits") + export_notes: list[dict[str, Any]] = [] + + for exp in exports or []: + ep = Path(exp) + export_notes.append({"path": str(ep.resolve()) if ep.exists() else str(ep), "name": ep.name}) + extra = inspect_export(ep) + for area_id, rows in extra.items(): + if area_id in AREA_SPECS: + combined.setdefault(area_id, []).extend( + [{**row, "root": ep.name} for row in rows] + ) + elif area_id in SPECIALIST_SPECS: + combined_spec.setdefault(area_id, []).extend( + [{**row, "root": ep.name} for row in rows] + ) + + focus: list[dict[str, Any]] = [] + for area_id, spec in AREA_SPECS.items(): + entry = _area_entry(area_id, spec, combined.get(area_id) or []) + if entry: + focus.append(entry) + for area_id, spec in SPECIALIST_SPECS.items(): + entry = _area_entry(area_id, spec, combined_spec.get(area_id) or []) + if entry: + focus.append(entry) + focus.sort(key=lambda a: (-int(a["signal"]), a["id"])) + for i, area in enumerate(focus, start=1): + area["rank"] = i + + ecosystems: list[str] = [] + cicd_handoff: list[dict[str, Any]] = [] + for info in inspected: + for eco in info["ecosystems"]: + if eco not in ecosystems: + ecosystems.append(eco) + cicd_handoff.extend(info.get("cicd_handoff") or []) + + slug = slugify(system_name) + return { + "version": PLAN_VERSION, + "system": system_name, + "system_slug": slug, + "roots": [ + { + "path": i["path"], + "name": i["name"], + "top_level": i["top_level"], + "major_dirs": i["major_dirs"], + "ecosystems": i["ecosystems"], + "layout": i["layout"], + "file_count": i["file_count"], + } + for i in inspected + ], + "exports": export_notes, + "ecosystems": ecosystems, + "focus_areas": focus, + "specialists": [ + { + "id": a["id"], + "kind": a.get("kind"), + "parent": a.get("parent"), + "agent": a.get("agent"), + "title": a.get("title"), + "signal": a.get("signal"), + "hit_count": a.get("hit_count"), + } + for a in focus + if a.get("kind") in ("orchestration", "elt-tool") + ], + "cicd_handoff": cicd_handoff[:MAX_HITS], + "artifacts": { + "plan_json": f"{DEKC_DIRNAME}/{PLAN_JSON_NAME}", + "plan_md": f"{DEKC_DIRNAME}/{PLAN_MD_NAME}", + "progress": f"{DEKC_DIRNAME}/{PLAN_PROGRESS_NAME}", + }, + } + + +def empty_progress(plan: dict[str, Any]) -> dict[str, Any]: + areas: dict[str, Any] = {} + for area in plan.get("focus_areas") or []: + areas[area["id"]] = { + "items": { + item["id"]: {"status": item.get("status") or "pending", "note": item.get("note") or ""} + for item in area.get("checklist") or [] + } + } + return { + "plan": plan.get("artifacts", {}).get("plan_json", f"{DEKC_DIRNAME}/{PLAN_JSON_NAME}"), + "areas": areas, + } + + +def apply_progress(plan: dict[str, Any], progress: dict[str, Any]) -> dict[str, Any]: + areas = progress.get("areas") or {} + for area in plan.get("focus_areas") or []: + rec = areas.get(area["id"]) or {} + items = rec.get("items") or {} + for item in area.get("checklist") or []: + st = items.get(item["id"]) or {} + if st.get("status"): + item["status"] = st["status"] + if "note" in st: + item["note"] = st.get("note") or "" + return plan + + +def checklist_summary(plan: dict[str, Any]) -> dict[str, int]: + done = blocked = pending = total = 0 + for area in plan.get("focus_areas") or []: + for item in area.get("checklist") or []: + total += 1 + status = item.get("status") or "pending" + if status == "done": + done += 1 + elif status == "blocked": + blocked += 1 + else: + pending += 1 + return {"done": done, "blocked": blocked, "pending": pending, "total": total} + + +def scan_domains_from_plan(plan: dict[str, Any], *, area: str | None = None) -> list[str]: + domains: list[str] = [] + for focus in plan.get("focus_areas") or []: + if area and focus["id"] != area: + continue + for d in focus.get("scan_domains") or []: + if d not in domains: + domains.append(d) + return domains + + +def render_plan_markdown(plan: dict[str, Any]) -> str: + system = plan.get("system") or "Data platform" + lines: list[str] = [ + f"# Reverse-engineering plan: {system}", + "", + "Generated by `dekc_plan.py` (deterministic; presence and counts only).", + "Walkers populate the graph. Query-time `data-retriever` / `dekc-retrieve` stay separate.", + "Adversarial skeptics + `re-adversary-judge` still close every walk — this plan does not replace them.", + "", + "## Scan map", + "", + ] + for root in plan.get("roots") or []: + lines.append(f"### `{root.get('name')}` (`{root.get('path')}`)") + lines.append("") + lines.append(f"- Layout: **{root.get('layout')}**") + ecos = ", ".join(root.get("ecosystems") or []) or "_none detected_" + lines.append(f"- Ecosystems: {ecos}") + top = ", ".join(f"`{t}`" for t in (root.get("top_level") or [])) + if top: + lines.append(f"- Top-level: {top}") + majors = ", ".join(f"`{d}/`" for d in (root.get("major_dirs") or [])) + if majors: + lines.append(f"- Major dirs: {majors}") + lines.append(f"- Files visited (BFS cap): {root.get('file_count')}") + lines.append("") + + exports = plan.get("exports") or [] + if exports: + lines.extend(["## Export paths (control-plane JSON, no live APIs)", ""]) + for exp in exports: + lines.append(f"- `{exp.get('name')}` — `{exp.get('path')}`") + lines.append("") + + def _table(areas: list[dict[str, Any]]) -> None: + lines.extend( + [ + "| Rank | Area | Signal | Hits | Agent | Scan domains |", + "|------|------|--------|------|-------|--------------|", + ] + ) + for area in areas: + domains = ", ".join(area.get("scan_domains") or []) or "enrichment only" + lines.append( + f"| {area.get('rank')} | `{area['id']}` — {area.get('title')} | " + f"{area.get('signal')} | {area.get('hit_count')} | `{area.get('agent')}` | `{domains}` |" + ) + lines.append("") + + domains_only = [a for a in (plan.get("focus_areas") or []) if a.get("kind", "domain") == "domain"] + orch_specs = [a for a in (plan.get("focus_areas") or []) if a.get("kind") == "orchestration"] + elt_specs = [a for a in (plan.get("focus_areas") or []) if a.get("kind") == "elt-tool"] + + lines.extend(["## Focus areas (ranked by signal)", ""]) + if domains_only: + _table(domains_only) + else: + lines.append("_None — no lake / ELT / orchestration signals._") + lines.append("") + + lines.extend( + [ + "## Orchestration specialists (signal-gated)", + "", + "Spawn **only** when markers exist. No Airflow scout without DAG / `airflow` markers.", + "No Glue scout without `awsglue` / Glue job scripts. Specialists enrich after the", + "deterministic walk — they do not replace `dekc_walk.py`.", + "", + ] + ) + if orch_specs: + _table(orch_specs) + else: + lines.append("_None — no orchestration specialist signals._") + lines.append("") + + lines.extend( + [ + "## ELT specialists (signal-gated)", + "", + "Spawn **only** when markers exist (no DuckDB scout without `*.duckdb` / duckdb SQL;", + "no notebook scout without `.ipynb` or Fabric notebook export).", + "", + ] + ) + if elt_specs: + _table(elt_specs) + else: + lines.append("_None — no ELT specialist signals._") + lines.append("") + + handoff = plan.get("cicd_handoff") or [] + lines.extend( + [ + "## CI/CD boundary (SAC owns pipelines)", + "", + "DEKC captures **data jobs** (`IngestionJob`). SAC captures **CI/CD pipelines**.", + "If Actions only trigger Glue/dbt, cross-link the workflow; do not steal SAC Pipeline nouns.", + "", + ] + ) + if handoff: + sample = ", ".join(f"`{h.get('path')}`" for h in handoff[:8]) + lines.append(f"Handoff signals: {sample}") + lines.append("") + else: + lines.append("_No CI files that mention Glue/dbt/Airflow were seen._") + lines.append("") + + lines.extend( + [ + "## Suggested fan-out", + "", + "Parent reviews this plan, then **spawns one child per domain area and each listed specialist**.", + "Independent domains run in parallel. Then existing **adversarial skeptics** → `re-adversary-judge`.", + "Do **not** spawn a specialist the plan did not list. Retrievers are not walkers.", + "", + "```bash", + "# Pause after plan", + 'python3 scripts/dekc_orchestrate.py --plan-only --system "Name" --scan-root ', + "# Child: one area, domain-scoped walk + capture", + "python3 scripts/dekc_orchestrate.py --from-plan knowledge/.dekc/re-plan.json --area lake \\", + ' --system "Name" --scan-root ', + "python3 scripts/dekc_plan.py mark --plan knowledge/.dekc/re-plan.json --area lake \\", + " --item inventory --status done", + "# After producers: grade + skeptics + judge (unchanged)", + "python3 scripts/dekc_grade.py --repo . --bundle knowledge", + "```", + "", + "## Deep-dive checklists", + "", + ] + ) + for area in plan.get("focus_areas") or []: + kind = area.get("kind") or "domain" + lines.append(f"### `{area['id']}` — `{area.get('agent')}` ({kind})") + lines.append("") + lines.append(area.get("title") or area["id"]) + sample = ", ".join(f"`{h.get('path')}`" for h in (area.get("hits") or [])[:8] if h.get("path")) + if sample: + lines.append("") + lines.append(f"Signals: {sample}") + lines.append("") + for item in area.get("checklist") or []: + status = item.get("status") or "pending" + box = "x" if status == "done" else " " + note = item.get("note") or "" + extra = "" + if status == "blocked": + extra = f" *(blocked{': ' + note if note else ''})*" + elif note and status == "done": + extra = f" *({note})*" + lines.append(f"- [{box}] `{item['id']}` — {item['text']}{extra}") + lines.append("") + + summary = checklist_summary(plan) + lines.extend( + [ + "## Checklist summary", + "", + f"- done: {summary['done']}", + f"- blocked: {summary['blocked']}", + f"- pending: {summary['pending']}", + f"- total: {summary['total']}", + "", + ] + ) + return "\n".join(lines) + + +def write_plan( + bundle: Path, + roots: list[Path], + *, + system_name: str, + exports: list[Path] | None = None, + progress: dict[str, Any] | None = None, +) -> dict[str, Any]: + bundle = Path(bundle) + plan = build_plan(roots, system_name=system_name, exports=exports) + if progress: + apply_progress(plan, progress) + prog = progress + else: + prog = empty_progress(plan) + paths = plan_paths(bundle) + paths["dir"].mkdir(parents=True, exist_ok=True) + paths["json"].write_text(json.dumps(plan, indent=2) + "\n", encoding="utf-8") + paths["md"].write_text(render_plan_markdown(plan), encoding="utf-8") + paths["progress"].write_text(json.dumps(prog, indent=2) + "\n", encoding="utf-8") + plan = dict(plan) + plan["written"] = {k: str(v) for k, v in paths.items()} + plan["checklist"] = checklist_summary(plan) + return plan + + +def load_plan(plan_or_bundle: Path) -> dict[str, Any]: + files = resolve_plan_files(plan_or_bundle) + if not files["json"].is_file(): + raise FileNotFoundError(f"RE plan not found: {files['json']}") + plan = json.loads(files["json"].read_text(encoding="utf-8")) + if files["progress"].is_file(): + prog = json.loads(files["progress"].read_text(encoding="utf-8")) + apply_progress(plan, prog) + return plan + + +def mark_checklist( + plan_or_bundle: Path, + *, + area: str, + item: str, + status: str, + note: str = "", +) -> dict[str, Any]: + if status not in {"pending", "done", "blocked"}: + raise ValueError(f"status must be pending|done|blocked, got {status!r}") + files = resolve_plan_files(plan_or_bundle) + plan = load_plan(files["json"]) + focus_ids = {a["id"] for a in plan.get("focus_areas") or []} + if area not in focus_ids: + raise KeyError(f"area {area!r} is not in the plan ({sorted(focus_ids)})") + item_ids: list[str] = [] + for a in plan["focus_areas"]: + if a["id"] == area: + item_ids = [i["id"] for i in a.get("checklist") or []] + break + if item not in item_ids: + raise KeyError(f"item {item!r} is not in area {area!r} ({item_ids})") + if files["progress"].is_file(): + prog = json.loads(files["progress"].read_text(encoding="utf-8")) + else: + prog = empty_progress(plan) + areas = prog.setdefault("areas", {}) + rec = areas.setdefault(area, {"items": {}}) + items = rec.setdefault("items", {}) + items[item] = {"status": status, "note": note} + apply_progress(plan, prog) + files["dir"].mkdir(parents=True, exist_ok=True) + files["json"].write_text(json.dumps(plan, indent=2) + "\n", encoding="utf-8") + files["progress"].write_text(json.dumps(prog, indent=2) + "\n", encoding="utf-8") + files["md"].write_text(render_plan_markdown(plan), encoding="utf-8") + return { + "area": area, + "item": item, + "status": status, + "note": note, + "checklist": checklist_summary(plan), + "written": {k: str(v) for k, v in files.items()}, + } + + +def mark_area_item_if_present( + plan_or_bundle: Path, + *, + area: str, + item: str, + status: str = "done", + note: str = "", +) -> dict[str, Any] | None: + try: + return mark_checklist(plan_or_bundle, area=area, item=item, status=status, note=note) + except (FileNotFoundError, KeyError): + return None + + +def main(argv: list[str] | None = None) -> int: + p = argparse.ArgumentParser(description="DEKC breadth-first reverse-engineering plan") + sub = p.add_subparsers(dest="cmd") + + p.add_argument("--repo", default=".", help="Knowledge host repo") + p.add_argument("--bundle", default=None) + p.add_argument("--system", default="Data platform") + p.add_argument("--scan-root", action="append", default=[], help="Lake / SQL / job root(s) to map") + p.add_argument( + "--export", + action="append", + default=[], + help="Optional Fabric / PBI / inventory JSON (no live APIs)", + ) + p.add_argument("--json", action="store_true") + p.add_argument("--write", action="store_true", help="Write .dekc/re-plan.{json,md} into the bundle") + + p_mark = sub.add_parser("mark", help="Check off (or block) a deep-dive checklist item") + p_mark.add_argument("--plan", default=None, help="Plan JSON, .dekc dir, or bundle") + p_mark.add_argument("--repo", default=".") + p_mark.add_argument("--bundle", default=None) + p_mark.add_argument("--area", required=True) + p_mark.add_argument("--item", required=True) + p_mark.add_argument("--status", default="done", choices=("pending", "done", "blocked")) + p_mark.add_argument("--note", default="") + p_mark.add_argument("--json", action="store_true") + + p_show = sub.add_parser("show", help="Print an existing plan") + p_show.add_argument("--plan", default=None) + p_show.add_argument("--repo", default=".") + p_show.add_argument("--bundle", default=None) + p_show.add_argument("--json", action="store_true") + + args = p.parse_args(argv) + from dekc_common import resolve_knowledge_root + + if args.cmd == "mark": + host = Path(args.repo).resolve() + target = Path(args.plan).resolve() if args.plan else resolve_knowledge_root(host, args.bundle) + result = mark_checklist(target, area=args.area, item=args.item, status=args.status, note=args.note) + if args.json: + print(json.dumps(result, indent=2)) + else: + print(f"marked {args.area}/{args.item} -> {args.status}") + print(f"checklist: {result['checklist']}") + return 0 + + if args.cmd == "show": + host = Path(args.repo).resolve() + target = Path(args.plan).resolve() if args.plan else resolve_knowledge_root(host, args.bundle) + plan = load_plan(target) + if args.json: + print(json.dumps(plan, indent=2, default=str)) + else: + print(render_plan_markdown(plan)) + return 0 + + host = Path(args.repo).resolve() + roots = [Path(r).resolve() for r in (args.scan_root or [str(host)])] + exports = [Path(e).resolve() for e in (args.export or [])] + if args.write: + bundle = resolve_knowledge_root(host, args.bundle) + bundle.mkdir(parents=True, exist_ok=True) + plan = write_plan(bundle, roots, system_name=args.system, exports=exports) + else: + plan = build_plan(roots, system_name=args.system, exports=exports) + plan["checklist"] = checklist_summary(plan) + if args.json: + print(json.dumps(plan, indent=2, default=str)) + else: + print(render_plan_markdown(plan)) + if args.write: + written = plan.get("written") or {} + print(f"\nWrote {written.get('md')} and {written.get('json')}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/dekc_walk.py b/scripts/dekc_walk.py index ea18f43..1c420c6 100755 --- a/scripts/dekc_walk.py +++ b/scripts/dekc_walk.py @@ -8,8 +8,11 @@ - Medallion folder conventions: bronze/, silver/, gold/, raw/ - dbt models (models/**/*.sql + schema.yml) - Spark/Delta path markers (_delta_log, *.parquet dir names) + - DuckDB files (*.duckdb) and notebook SQL (*.ipynb) + - Orchestration / ELT / DQ markers (Airflow, Glue, ADF, dbt, GE/Soda) Agents orchestrate this script; subagents specialize on schema, lineage, semantic. +Cheap discovery writes live here. `dekc_plan.py` is presence-only. """ from __future__ import annotations @@ -44,7 +47,14 @@ utc_now, write_knowledge, ) -from dekc_platform import capture_data_lake, capture_ingestion_job, capture_stream # noqa: E402 +from dekc_platform import ( # noqa: E402 + capture_data_catalog, + capture_data_lake, + capture_dq_rule, + capture_ingestion_job, + capture_storage, + capture_stream, +) SQL_FROM_RE = re.compile( r"\b(?:from|join)\s+([`\"\[]?[\w.-]+[`\"\]]?(?:\.[`\"\[]?[\w.-]+[`\"\]]?){0,2})", @@ -251,6 +261,36 @@ def walk_lake( ): result.record(rel, action) + # Delta table directories (_delta_log sibling of parquet) + seen_delta: set[str] = set() + for delta_dir in lake_root.rglob("_delta_log"): + if not delta_dir.is_dir(): + continue + parent = delta_dir.parent + try: + key = str(parent.relative_to(lake_root)) + except ValueError: + continue + if key in seen_delta or key in seen_dirs: + continue + seen_delta.add(key) + tname = parent.name + if tname.startswith("_"): + continue + result.discovered["delta_tables"] = result.discovered.get("delta_tables", 0) + 1 + if dry_run: + result.skipped.append(key) + continue + layer = infer_layer(parent, lake_root) + for rel, action in capture_table( + bundle, + name=tname, + layer=layer if layer in ("bronze", "silver", "gold", "raw") else "bronze", + description=f"Delta table at {key}", + source=src_name, + ): + result.record(rel, action) + # Infer bronze→silver→gold promotions when same basename appears in multiple layers by_base: dict[str, list[str]] = {} for rel in result.created + result.updated + result.skipped: @@ -532,6 +572,504 @@ def walk_inventory_json( return result +def extract_notebook_sql(text_or_path: str | Path) -> list[str]: + """Pull SQL-ish cells from an .ipynb (%%sql, spark.sql, duckdb, SELECT).""" + if isinstance(text_or_path, Path): + try: + data = json.loads(text_or_path.read_text(encoding="utf-8", errors="replace")) + except (OSError, json.JSONDecodeError): + return [] + else: + try: + data = json.loads(text_or_path) + except json.JSONDecodeError: + return [] + if not isinstance(data, dict): + return [] + found: list[str] = [] + for cell in data.get("cells") or []: + if not isinstance(cell, dict) or cell.get("cell_type") != "code": + continue + src = cell.get("source") or "" + if isinstance(src, list): + src = "".join(src) + blob = str(src).strip() + if not blob: + continue + low = blob.lower() + if ( + "%%sql" in low + or "spark.sql" in low + or "duckdb.sql" in low + or "duckdb.execute" in low + or re.search(r"\bselect\b", blob, re.IGNORECASE) + ): + found.append(blob) + return found + + +def _job_name_from_file(path: Path, root: Path, text: str = "") -> str: + m = re.search(r"dag_id\s*=\s*['\"]([^'\"]+)", text) + if m: + return m.group(1) + m = re.search(r"expectation_suite_name['\"]\s*:\s*['\"]([^'\"]+)", text) + if m: + return m.group(1) + return path.stem + + +def walk_duckdb( + lake_root: Path, + bundle: Path, + *, + dry_run: bool = False, +) -> WalkResult: + result = WalkResult() + files = list(lake_root.rglob("*.duckdb")) + list(lake_root.rglob("*.ddb")) + result.discovered["duckdb_files"] = len(files) + for f in files[:200]: + name = f.stem + if dry_run: + result.skipped.append(str(f)) + continue + for rel, action in capture_source( + bundle, + name=name, + kind="duckdb", + uri=str(f.resolve()), + description=f"DuckDB file at {f.relative_to(lake_root)} (binary not opened).", + ): + result.record(rel, action) + return result + + +def walk_notebooks( + lake_root: Path, + bundle: Path, + *, + dry_run: bool = False, + max_files: int = 80, +) -> WalkResult: + result = WalkResult() + nbs = list(lake_root.rglob("*.ipynb"))[:max_files] + result.discovered["notebooks"] = len(nbs) + for nb in nbs: + sqls = extract_notebook_sql(nb) + if dry_run: + result.skipped.append(str(nb)) + continue + if not sqls: + for rel, action in capture_query( + bundle, + name=nb.stem, + dialect="notebook", + body_sql="", + description=f"Notebook {nb.relative_to(lake_root)} (no SQL cells extracted).", + ): + result.record(rel, action) + continue + for i, sql in enumerate(sqls): + qname = nb.stem if len(sqls) == 1 else f"{nb.stem}-cell-{i + 1}" + refs = extract_sql_tables(sql) + for rel, action in capture_query( + bundle, + name=qname, + dialect="sql", + body_sql=sql, + description=f"SQL cell from {nb.relative_to(lake_root)}", + reads_from=refs, + ): + result.record(rel, action) + return result + + +def walk_dq_markers( + lake_root: Path, + bundle: Path, + *, + dry_run: bool = False, + max_files: int = 80, +) -> WalkResult: + """Cheap DQRule capture from GE / Soda / dbt test files. No runtime.""" + result = WalkResult() + candidates: list[Path] = [] + for pat in ( + "**/great_expectations.yml", + "**/great_expectations.yaml", + "**/expectations/*.json", + "**/soda*.yml", + "**/soda*.yaml", + "**/checks.yml", + "**/schema.yml", + ): + candidates.extend(lake_root.glob(pat)) + seen: set[str] = set() + for f in candidates: + if not f.is_file(): + continue + key = str(f) + if key in seen: + continue + seen.add(key) + if len(seen) > max_files: + break + text = f.read_text(encoding="utf-8", errors="replace") + low = text.lower() + name = f.stem + rule_type = "expectation" + if "great_expectation" in str(f).lower() or "expectation_suite" in low: + rule_type = "great-expectations" + suite = _job_name_from_file(f, lake_root, text) + name = suite or name + elif "soda" in str(f).lower() or "checks for" in low: + rule_type = "soda" + elif "tests:" in text or "data_tests:" in text: + if "dbt" not in str(f).lower() and "models" not in f.parts: + continue + rule_type = "dbt-test" + else: + if f.name.startswith("great_expectations"): + rule_type = "great-expectations" + elif f.name.startswith("soda") or f.name == "checks.yml": + rule_type = "soda" + else: + continue + result.discovered["dq_files"] = result.discovered.get("dq_files", 0) + 1 + if dry_run: + result.skipped.append(str(f)) + continue + expr = "" + if '"expectation_type"' in text: + m = re.search(r'"expectation_type"\s*:\s*"([^"]+)"', text) + if m: + expr = m.group(1) + for rel, action in capture_dq_rule( + bundle, + name=name, + description=f"DQ marker from {f.relative_to(lake_root)} (not executed).", + rule_type=rule_type, + expression=expr, + ): + result.record(rel, action) + return result + + +def walk_elt( + lake_root: Path, + bundle: Path, + *, + dry_run: bool = False, +) -> WalkResult: + result = WalkResult() + for proj in list(lake_root.rglob("dbt_project.yml")) + list(lake_root.rglob("dbt_project.yaml")): + result.discovered["dbt_projects"] = result.discovered.get("dbt_projects", 0) + 1 + if dry_run: + result.skipped.append(str(proj)) + continue + name = proj.parent.name + for rel, action in capture_ingestion_job( + bundle, + name=f"dbt-{name}", + description=f"dbt project at {proj.relative_to(lake_root)}", + orchestrator="dbt", + mode="batch", + target_layer="", + ): + result.record(rel, action) + spark_hits = 0 + for py in lake_root.rglob("*.py"): + if spark_hits >= 40: + break + try: + text = py.read_text(encoding="utf-8", errors="replace")[:4000] + except OSError: + continue + if "SparkSession" in text or "spark-submit" in text: + if "awsglue" in text: + continue + spark_hits += 1 + if dry_run: + result.skipped.append(str(py)) + continue + for rel, action in capture_ingestion_job( + bundle, + name=py.stem, + description=f"Spark/EMR job script {py.relative_to(lake_root)}", + orchestrator="spark", + mode="batch", + target_layer="", + ): + result.record(rel, action) + result.discovered["spark_jobs"] = spark_hits + return result + + +def walk_orchestration( + lake_root: Path, + bundle: Path, + *, + dry_run: bool = False, + kinds: set[str] | None = None, +) -> WalkResult: + """Cheap IngestionJob capture for evidenced orchestrators. No live APIs.""" + result = WalkResult() + want = kinds or { + "airflow", + "glue", + "adf", + "stepfunctions", + "composer", + "cron", + "fabric", + } + + if "airflow" in want: + for py in lake_root.rglob("*.py"): + parts = {p.lower() for p in py.parts} + try: + text = py.read_text(encoding="utf-8", errors="replace")[:4000] + except OSError: + continue + if not ( + "dags" in parts + or "from airflow" in text + or "import airflow" in text + or "DAG(" in text + ): + continue + result.discovered["airflow_dags"] = result.discovered.get("airflow_dags", 0) + 1 + if dry_run: + result.skipped.append(str(py)) + continue + for rel, action in capture_ingestion_job( + bundle, + name=_job_name_from_file(py, lake_root, text), + description=f"Airflow DAG {py.relative_to(lake_root)}", + orchestrator="airflow", + mode="batch", + target_layer="", + ): + result.record(rel, action) + + if "glue" in want: + for py in lake_root.rglob("*.py"): + try: + text = py.read_text(encoding="utf-8", errors="replace")[:4000] + except OSError: + continue + if "awsglue" not in text and "GlueContext" not in text: + continue + result.discovered["glue_jobs"] = result.discovered.get("glue_jobs", 0) + 1 + if dry_run: + result.skipped.append(str(py)) + continue + for rel, action in capture_ingestion_job( + bundle, + name=py.stem, + description=f"Glue job script {py.relative_to(lake_root)}", + orchestrator="glue", + mode="batch", + target_layer="", + ): + result.record(rel, action) + + if "stepfunctions" in want: + for jf in list(lake_root.rglob("*.json")) + list(lake_root.rglob("*.asl.json")): + try: + text = jf.read_text(encoding="utf-8", errors="replace")[:4000] + except OSError: + continue + if '"StartAt"' not in text or '"States"' not in text: + continue + result.discovered["stepfunctions"] = result.discovered.get("stepfunctions", 0) + 1 + if dry_run: + result.skipped.append(str(jf)) + continue + for rel, action in capture_ingestion_job( + bundle, + name=jf.stem.replace(".asl", ""), + description=f"Step Functions ASL {jf.relative_to(lake_root)}", + orchestrator="stepfunctions", + mode="batch", + target_layer="", + ): + result.record(rel, action) + + if "adf" in want: + for jf in list(lake_root.rglob("*.json")) + list(lake_root.rglob("*.yml")): + try: + text = jf.read_text(encoding="utf-8", errors="replace")[:4000] + except OSError: + continue + if "Microsoft.DataFactory" not in text: + continue + result.discovered["adf_pipelines"] = result.discovered.get("adf_pipelines", 0) + 1 + if dry_run: + result.skipped.append(str(jf)) + continue + for rel, action in capture_ingestion_job( + bundle, + name=jf.stem, + description=f"ADF pipeline {jf.relative_to(lake_root)}", + orchestrator="adf", + mode="batch", + target_layer="", + ): + result.record(rel, action) + + if "composer" in want: + for yf in list(lake_root.rglob("*.yml")) + list(lake_root.rglob("*.yaml")): + try: + text = yf.read_text(encoding="utf-8", errors="replace")[:4000] + except OSError: + continue + if "composer.googleapis.com" not in text and "composer" not in {p.lower() for p in yf.parts}: + continue + if "composer.googleapis.com" not in text and "airflowConfigOverrides" not in text: + continue + result.discovered["composer"] = result.discovered.get("composer", 0) + 1 + if dry_run: + result.skipped.append(str(yf)) + continue + for rel, action in capture_ingestion_job( + bundle, + name=yf.parent.name or yf.stem, + description=f"Cloud Composer marker {yf.relative_to(lake_root)}", + orchestrator="composer", + mode="batch", + target_layer="", + ): + result.record(rel, action) + + if "cron" in want: + for yf in list(lake_root.rglob("*.yml")) + list(lake_root.rglob("*.yaml")): + try: + text = yf.read_text(encoding="utf-8", errors="replace")[:4000] + except OSError: + continue + if "kind: CronJob" not in text: + continue + low = text.lower() + if not any(n in low for n in ("dbt", "spark-submit", "python", "glue", "airflow")): + continue + result.discovered["cronjobs"] = result.discovered.get("cronjobs", 0) + 1 + if dry_run: + result.skipped.append(str(yf)) + continue + for rel, action in capture_ingestion_job( + bundle, + name=yf.stem, + description=f"K8s CronJob loader {yf.relative_to(lake_root)}", + orchestrator="k8s-cronjob", + mode="batch", + schedule="", + target_layer="", + ): + result.record(rel, action) + + return result + + +def walk_storage_markers( + lake_root: Path, + bundle: Path, + *, + dry_run: bool = False, +) -> WalkResult: + result = WalkResult() + for hint in ("storage", "s3", "adls"): + for d in lake_root.rglob(hint): + if not d.is_dir(): + continue + result.discovered["storage_dirs"] = result.discovered.get("storage_dirs", 0) + 1 + if dry_run: + result.skipped.append(str(d)) + continue + for rel, action in capture_storage( + bundle, + name=d.name, + kind="prefix", + uri=str(d.resolve()), + description=f"Storage path convention {d.relative_to(lake_root)}", + ): + result.record(rel, action) + catalogs = list(lake_root.rglob("catalog.json"))[:8] + for c in catalogs: + result.discovered["catalog_json"] = result.discovered.get("catalog_json", 0) + 1 + if dry_run: + result.skipped.append(str(c)) + continue + for rel, action in capture_data_catalog( + bundle, + name=c.parent.name or c.stem, + description=f"Catalog export {c.relative_to(lake_root)}", + ): + result.record(rel, action) + return result + + +def walk_scoped( + lake_root: Path | None, + bundle: Path, + *, + domains: list[str], + source_name: str | None = None, + fabric_items: Path | None = None, + pbi_bindings: Path | None = None, + inventory: Path | None = None, + workspace: str = "", + workspace_id: str = "", + inventory_layer: str = "gold", + max_files: int = 500, + dry_run: bool = False, +) -> WalkResult: + """Domain-scoped capture used by `--from-plan --area` / orchestrate.""" + result = WalkResult() + wanted = set(domains or []) + if lake_root and "lake" in wanted: + _merge_walk( + result, + walk_lake( + lake_root, + bundle, + source_name=source_name, + max_files=max_files, + dry_run=dry_run, + ), + ) + if lake_root and "elt" in wanted: + _merge_walk(result, walk_elt(lake_root, bundle, dry_run=dry_run)) + if lake_root and "orchestration" in wanted: + _merge_walk(result, walk_orchestration(lake_root, bundle, dry_run=dry_run)) + if lake_root and "catalogs" in wanted: + _merge_walk(result, walk_storage_markers(lake_root, bundle, dry_run=dry_run)) + if inventory and "catalogs" in wanted: + _merge_walk( + result, + walk_inventory_json(inventory, bundle, layer=inventory_layer, dry_run=dry_run), + ) + if pbi_bindings and "bi" in wanted: + _merge_walk(result, walk_pbi_bindings(pbi_bindings, bundle, dry_run=dry_run)) + if fabric_items and wanted.intersection({"bi", "orchestration", "orch-fabric", "notebooks"}): + _merge_walk( + result, + walk_fabric_items( + fabric_items, + bundle, + workspace=workspace, + workspace_id=workspace_id, + dry_run=dry_run, + ), + ) + if lake_root and "dq" in wanted: + _merge_walk(result, walk_dq_markers(lake_root, bundle, dry_run=dry_run)) + if lake_root and "duckdb" in wanted: + _merge_walk(result, walk_duckdb(lake_root, bundle, dry_run=dry_run)) + if lake_root and "notebooks" in wanted: + _merge_walk(result, walk_notebooks(lake_root, bundle, dry_run=dry_run)) + return result + + def _merge_walk(into: WalkResult, other: WalkResult) -> WalkResult: into.created.extend(other.created) into.updated.extend(other.updated) @@ -569,52 +1107,132 @@ def main(argv: list[str] | None = None) -> int: parser.add_argument("--workspace", default="") parser.add_argument("--workspace-id", default="") parser.add_argument("--inventory-layer", default="gold") + parser.add_argument("--system", default="Data platform", help="With --plan-only: system name") + parser.add_argument( + "--plan-only", + action="store_true", + help="Write a breadth-first RE plan for the path, then stop (no capture)", + ) + parser.add_argument( + "--from-plan", + default=None, + help="Existing .dekc/re-plan.json — scope capture with --area", + ) + parser.add_argument( + "--area", + default=None, + help="With --from-plan: one focus area (domain-scoped capture)", + ) + parser.add_argument( + "--export", + action="append", + default=[], + help="Optional export JSON for --plan-only (Fabric / inventory / PBI)", + ) args = parser.parse_args(argv) from dekc_common import resolve_author + + repo = Path(args.repo).resolve() + bundle = resolve_knowledge_root(repo, args.bundle) + lake = Path(args.path).resolve() if args.path else None + + if args.plan_only: + from dekc_plan import write_plan + + roots = [lake] if lake else [repo] + exports = [Path(e).resolve() for e in (args.export or [])] + if args.fabric_items: + exports.append(Path(args.fabric_items).resolve()) + if args.inventory: + exports.append(Path(args.inventory).resolve()) + if args.pbi_bindings: + exports.append(Path(args.pbi_bindings).resolve()) + bundle.mkdir(parents=True, exist_ok=True) + plan = write_plan(bundle, roots, system_name=args.system, exports=exports) + if args.json: + print(json.dumps(plan, indent=2, default=str)) + else: + from dekc_plan import render_plan_markdown + + print(render_plan_markdown(plan)) + written = plan.get("written") or {} + print(f"\nWrote {written.get('md')} and {written.get('json')}") + return 0 + if not args.dry_run: resolve_author(args.author) - if not args.path and not args.fabric_items and not args.pbi_bindings and not args.inventory: - parser.error("provide a filesystem path and/or --fabric-items / --pbi-bindings / --inventory") + if args.from_plan: + from dekc_plan import load_plan, scan_domains_from_plan - repo = Path(args.repo).resolve() - bundle = resolve_knowledge_root(repo, args.bundle) - ensure_bundle(bundle) - result = WalkResult() - if args.path: - result = walk_lake( - Path(args.path).resolve(), + plan = load_plan(Path(args.from_plan).resolve() if args.from_plan else bundle) + domains = scan_domains_from_plan(plan, area=args.area) + if args.area and not domains: + payload = { + "enrichment_only": True, + "area": args.area, + "domains": [], + "counts": {"created": 0, "updated": 0, "skipped": 0, "errors": 0}, + } + print(json.dumps(payload, indent=2) if args.json else f"enrichment only: area={args.area}") + return 0 + if not args.path and not args.fabric_items and not args.pbi_bindings and not args.inventory: + parser.error("provide a filesystem path and/or --fabric-items / --pbi-bindings / --inventory") + ensure_bundle(bundle) + result = walk_scoped( + lake, bundle, + domains=domains or ["lake"], source_name=args.source_name, + fabric_items=Path(args.fabric_items).resolve() if args.fabric_items else None, + pbi_bindings=Path(args.pbi_bindings).resolve() if args.pbi_bindings else None, + inventory=Path(args.inventory).resolve() if args.inventory else None, + workspace=args.workspace, + workspace_id=args.workspace_id, + inventory_layer=args.inventory_layer, max_files=args.max_files, dry_run=args.dry_run, ) - if args.fabric_items: - _merge_walk( - result, - walk_fabric_items( - Path(args.fabric_items).resolve(), - bundle, - workspace=args.workspace, - workspace_id=args.workspace_id, - dry_run=args.dry_run, - ), - ) - if args.pbi_bindings: - _merge_walk( - result, - walk_pbi_bindings(Path(args.pbi_bindings).resolve(), bundle, dry_run=args.dry_run), - ) - if args.inventory: - _merge_walk( - result, - walk_inventory_json( - Path(args.inventory).resolve(), + else: + if not args.path and not args.fabric_items and not args.pbi_bindings and not args.inventory: + parser.error("provide a filesystem path and/or --fabric-items / --pbi-bindings / --inventory") + + ensure_bundle(bundle) + result = WalkResult() + if args.path: + result = walk_lake( + Path(args.path).resolve(), bundle, - layer=args.inventory_layer, + source_name=args.source_name, + max_files=args.max_files, dry_run=args.dry_run, - ), - ) + ) + if args.fabric_items: + _merge_walk( + result, + walk_fabric_items( + Path(args.fabric_items).resolve(), + bundle, + workspace=args.workspace, + workspace_id=args.workspace_id, + dry_run=args.dry_run, + ), + ) + if args.pbi_bindings: + _merge_walk( + result, + walk_pbi_bindings(Path(args.pbi_bindings).resolve(), bundle, dry_run=args.dry_run), + ) + if args.inventory: + _merge_walk( + result, + walk_inventory_json( + Path(args.inventory).resolve(), + bundle, + layer=args.inventory_layer, + dry_run=args.dry_run, + ), + ) if (args.fabric_items or args.pbi_bindings or args.inventory) and not args.dry_run: n = len(result.created) + len(result.updated) append_log( diff --git a/skills/dekc-plan/SKILL.md b/skills/dekc-plan/SKILL.md new file mode 100644 index 0000000..cc2baab --- /dev/null +++ b/skills/dekc-plan/SKILL.md @@ -0,0 +1,39 @@ +--- +name: dekc-plan +description: Breadth-first reverse-engineering plan — map scan roots and optional exports, rank focus areas, emit deep-dive checklists. Pause here before spawning specialists. Not query-time retrieve. +--- + +# DEKC Plan + +Cheap, deterministic map of what is *present* in each `--scan-root` (plus optional Fabric / PBI / inventory `--export`). Presence and counts only — not a full walk and not `data-retriever`. + +```bash +python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_plan.py" \ + --repo . --system "$SYSTEM_NAME" --scan-root "$MIRROR" \ + --export workspace-items.json --write --json + +python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_orchestrate.py" \ + --repo . --system "$SYSTEM_NAME" --scan-root "$MIRROR" --plan-only --json + +# Same flags on walk +python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_walk.py" "$MIRROR" \ + --repo . --bundle knowledge --plan-only --system "$SYSTEM_NAME" +``` + +Writes `knowledge/.dekc/re-plan.md` + `.json` (+ progress) with ranked areas, **signal-gated** orchestration/ELT specialists, unchecked checklists, suggested sub-agents. Do **not** spawn a specialist the plan did not list (no Airflow scout without DAG markers). + +Mark progress after an area walker finishes an item: + +```bash +python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_plan.py" mark \ + --plan knowledge/.dekc/re-plan.json --area lake --item capture --status done +python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_plan.py" mark \ + --plan knowledge/.dekc/re-plan.json --area orch-airflow --item inventory --status done +python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_plan.py" mark \ + --plan knowledge/.dekc/re-plan.json --area dq --item capture --status blocked \ + --note "GE suite present; runtime not executed" +``` + +Then fan out with `reverse-engineering-orchestrator` / `--from-plan --area `. After producers: existing adversarial skeptics → `re-adversary-judge`. + +DEKC owns data orchestration + ELT/ETL. SAC owns CI/CD. If Actions only trigger Glue/dbt, cross-link — do not steal Pipeline nouns. diff --git a/skills/dekc-walk/SKILL.md b/skills/dekc-walk/SKILL.md index d983741..10d76d4 100644 --- a/skills/dekc-walk/SKILL.md +++ b/skills/dekc-walk/SKILL.md @@ -8,11 +8,21 @@ description: Walk a data lake/warehouse filesystem or ingest Fabric/Power BI inv `dekc_walk.py` is a **filesystem mirror walker** plus optional **control-plane JSON ingest**. It does not call Fabric REST itself. +Always **plan first** when reverse-engineering a new root (`/dekc-plan` or `--plan-only`). Then scoped capture: + ```bash # Git SQL / parquet mirror python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_walk.py" \ --repo . --bundle knowledge --source-name +# Pause after plan +python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_walk.py" \ + --repo . --bundle knowledge --plan-only --system "Retail Lake" + +# One plan area (not a full re-walk) +python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_walk.py" \ + --repo . --bundle knowledge --from-plan knowledge/.dekc/re-plan.json --area lake + # Fabric workspace items + Power BI bindings (export JSON first) python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_walk.py" \ --fabric-items workspace-items.json \ @@ -22,9 +32,12 @@ python3 "${CLAUDE_PLUGIN_ROOT}/scripts/dekc_walk.py" \ --repo . --bundle knowledge ``` -Then lineage + business promote + index (see data-lake-walker agent). +Or `dekc_orchestrate.py --from-plan … --area `. + +Then lineage + business promote + **adversarial grade** (see data-lake-walker / reverse-engineering-orchestrator). - `CREATE TABLE` with no `FROM` is reported as **DDL-only**, not “no lineage”. - Fabric `Report` captures as DEKC **Report**, not Dashboard. - Default SQL-endpoint SemanticModels are tagged in the description as not curated gold. - Grade a walk inside a mixed brain with `dekc_grade.py --prefix semantic,tables/gold-` (or `--tag`) rather than scoring 15k SAC nodes. +- DuckDB / notebooks / DQ / orchestration markers get cheap script writes when those plan areas run. Agents enrich; they do not invent edges. diff --git a/tests/fixtures/re-plan-lake/.github/workflows/run-glue.yml b/tests/fixtures/re-plan-lake/.github/workflows/run-glue.yml new file mode 100644 index 0000000..24d4f9e --- /dev/null +++ b/tests/fixtures/re-plan-lake/.github/workflows/run-glue.yml @@ -0,0 +1,9 @@ +name: trigger-glue +on: + schedule: + - cron: "0 6 * * *" +jobs: + run: + runs-on: ubuntu-latest + steps: + - run: echo "aws glue start-job-run --job-name orders_to_silver" diff --git a/tests/fixtures/re-plan-lake/adf/pipeline-copy-orders.json b/tests/fixtures/re-plan-lake/adf/pipeline-copy-orders.json new file mode 100644 index 0000000..03de4ea --- /dev/null +++ b/tests/fixtures/re-plan-lake/adf/pipeline-copy-orders.json @@ -0,0 +1,10 @@ +{ + "$schema": "https://schema.management.azure.com/schemas/2019-04-01/deploymentTemplate.json#", + "resources": [ + { + "type": "Microsoft.DataFactory/factories/pipelines", + "name": "lumenfield-copy-orders", + "properties": { "activities": [] } + } + ] +} diff --git a/tests/fixtures/re-plan-lake/analytics/local.duckdb b/tests/fixtures/re-plan-lake/analytics/local.duckdb new file mode 100644 index 0000000..e69de29 diff --git a/tests/fixtures/re-plan-lake/analytics/local_notes.sql b/tests/fixtures/re-plan-lake/analytics/local_notes.sql new file mode 100644 index 0000000..2d528ff --- /dev/null +++ b/tests/fixtures/re-plan-lake/analytics/local_notes.sql @@ -0,0 +1,2 @@ +-- duckdb local extract +SELECT * FROM read_parquet('lake/orders/*.parquet'); diff --git a/tests/fixtures/re-plan-lake/bronze/sales/orders_raw.sql b/tests/fixtures/re-plan-lake/bronze/sales/orders_raw.sql new file mode 100644 index 0000000..c1d2a1d --- /dev/null +++ b/tests/fixtures/re-plan-lake/bronze/sales/orders_raw.sql @@ -0,0 +1,2 @@ +CREATE TABLE bronze.orders_raw AS +SELECT * FROM landing.orders; diff --git a/tests/fixtures/re-plan-lake/composer/env.yaml b/tests/fixtures/re-plan-lake/composer/env.yaml new file mode 100644 index 0000000..afddde4 --- /dev/null +++ b/tests/fixtures/re-plan-lake/composer/env.yaml @@ -0,0 +1,3 @@ +# Google Cloud Composer environment (fiction) +airflowConfigOverrides: {} +# composer.googleapis.com/v1/projects/lumenfield/locations/us-central1/environments/orders diff --git a/tests/fixtures/re-plan-lake/dags/daily_orders.py b/tests/fixtures/re-plan-lake/dags/daily_orders.py new file mode 100644 index 0000000..76e5db6 --- /dev/null +++ b/tests/fixtures/re-plan-lake/dags/daily_orders.py @@ -0,0 +1,6 @@ +from airflow import DAG +from airflow.operators.bash import BashOperator +from datetime import datetime + +with DAG(dag_id="lumenfield_daily_orders", start_date=datetime(2026, 1, 1), schedule="@daily") as dag: + BashOperator(task_id="dbt_run", bash_command="dbt run --select orders") diff --git a/tests/fixtures/re-plan-lake/dbt/dbt_project.yml b/tests/fixtures/re-plan-lake/dbt/dbt_project.yml new file mode 100644 index 0000000..6f73a03 --- /dev/null +++ b/tests/fixtures/re-plan-lake/dbt/dbt_project.yml @@ -0,0 +1,3 @@ +name: lumenfield_orders +profile: lumenfield +model-paths: ["models"] diff --git a/tests/fixtures/re-plan-lake/dbt/models/schema.yml b/tests/fixtures/re-plan-lake/dbt/models/schema.yml new file mode 100644 index 0000000..0f8ff40 --- /dev/null +++ b/tests/fixtures/re-plan-lake/dbt/models/schema.yml @@ -0,0 +1,8 @@ +version: 2 +models: + - name: orders + tests: + - unique: + column_name: order_id + - not_null: + column_name: order_id diff --git a/tests/fixtures/re-plan-lake/dbt/models/silver/orders.sql b/tests/fixtures/re-plan-lake/dbt/models/silver/orders.sql new file mode 100644 index 0000000..e4d4559 --- /dev/null +++ b/tests/fixtures/re-plan-lake/dbt/models/silver/orders.sql @@ -0,0 +1 @@ +select * from {{ ref('orders_raw') }} diff --git a/tests/fixtures/re-plan-lake/exports/fabric-items.json b/tests/fixtures/re-plan-lake/exports/fabric-items.json new file mode 100644 index 0000000..d39a961 --- /dev/null +++ b/tests/fixtures/re-plan-lake/exports/fabric-items.json @@ -0,0 +1,8 @@ +{ + "value": [ + {"id": "pipe-1", "displayName": "LH Orders Pipeline", "type": "DataPipeline"}, + {"id": "nb-1", "displayName": "Clean Orders NB", "type": "Notebook"}, + {"id": "lh-1", "displayName": "RetailLake", "type": "Lakehouse"}, + {"id": "rpt-1", "displayName": "Exec Revenue", "type": "Report"} + ] +} diff --git a/tests/fixtures/re-plan-lake/exports/inventory.json b/tests/fixtures/re-plan-lake/exports/inventory.json new file mode 100644 index 0000000..148a797 --- /dev/null +++ b/tests/fixtures/re-plan-lake/exports/inventory.json @@ -0,0 +1,4 @@ +[ + {"TABLE_SCHEMA": "gold", "TABLE_NAME": "order_daily", "TABLE_TYPE": "BASE TABLE"}, + {"TABLE_SCHEMA": "gold", "TABLE_NAME": "customer_ltv", "TABLE_TYPE": "VIEW"} +] diff --git a/tests/fixtures/re-plan-lake/glue/jobs/orders_to_silver.py b/tests/fixtures/re-plan-lake/glue/jobs/orders_to_silver.py new file mode 100644 index 0000000..1eddd04 --- /dev/null +++ b/tests/fixtures/re-plan-lake/glue/jobs/orders_to_silver.py @@ -0,0 +1,5 @@ +from awsglue.context import GlueContext +from pyspark.context import SparkContext + +glue = GlueContext(SparkContext.getOrCreate()) +# lands bronze.orders_raw → silver.orders (fiction) diff --git a/tests/fixtures/re-plan-lake/gold/marts/order_daily.sql b/tests/fixtures/re-plan-lake/gold/marts/order_daily.sql new file mode 100644 index 0000000..74a2b96 --- /dev/null +++ b/tests/fixtures/re-plan-lake/gold/marts/order_daily.sql @@ -0,0 +1,2 @@ +CREATE TABLE gold.order_daily AS +SELECT order_date, SUM(amount) AS gmv FROM silver.orders GROUP BY 1; diff --git a/tests/fixtures/re-plan-lake/gx/expectations/orders_suite.json b/tests/fixtures/re-plan-lake/gx/expectations/orders_suite.json new file mode 100644 index 0000000..e7bff61 --- /dev/null +++ b/tests/fixtures/re-plan-lake/gx/expectations/orders_suite.json @@ -0,0 +1,6 @@ +{ + "expectation_suite_name": "lumenfield.orders", + "expectations": [ + {"expectation_type": "expect_column_values_to_not_be_null", "kwargs": {"column": "order_id"}} + ] +} diff --git a/tests/fixtures/re-plan-lake/gx/great_expectations.yml b/tests/fixtures/re-plan-lake/gx/great_expectations.yml new file mode 100644 index 0000000..3b9e7a5 --- /dev/null +++ b/tests/fixtures/re-plan-lake/gx/great_expectations.yml @@ -0,0 +1,2 @@ +config_version: 3 +datasources: {} diff --git a/tests/fixtures/re-plan-lake/k8s/nightly-dbt-cronjob.yaml b/tests/fixtures/re-plan-lake/k8s/nightly-dbt-cronjob.yaml new file mode 100644 index 0000000..cdf8dae --- /dev/null +++ b/tests/fixtures/re-plan-lake/k8s/nightly-dbt-cronjob.yaml @@ -0,0 +1,14 @@ +apiVersion: batch/v1 +kind: CronJob +metadata: + name: lumenfield-nightly-dbt +spec: + schedule: "15 6 * * *" + jobTemplate: + spec: + template: + spec: + containers: + - name: dbt + image: ghcr.io/lumenfield/dbt:latest + command: ["dbt", "run"] diff --git a/tests/fixtures/re-plan-lake/lake/orders/_delta_log/00000000000000000000.json b/tests/fixtures/re-plan-lake/lake/orders/_delta_log/00000000000000000000.json new file mode 100644 index 0000000..0967ef4 --- /dev/null +++ b/tests/fixtures/re-plan-lake/lake/orders/_delta_log/00000000000000000000.json @@ -0,0 +1 @@ +{} diff --git a/tests/fixtures/re-plan-lake/lake/orders/part-00000.parquet b/tests/fixtures/re-plan-lake/lake/orders/part-00000.parquet new file mode 100644 index 0000000..e69de29 diff --git a/tests/fixtures/re-plan-lake/notebooks/clean_orders.ipynb b/tests/fixtures/re-plan-lake/notebooks/clean_orders.ipynb new file mode 100644 index 0000000..451e3d0 --- /dev/null +++ b/tests/fixtures/re-plan-lake/notebooks/clean_orders.ipynb @@ -0,0 +1,38 @@ +{ + "nbformat": 4, + "nbformat_minor": 5, + "metadata": { + "kernelspec": { + "name": "python3", + "language": "python" + } + }, + "cells": [ + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "# Clean orders\n" + ] + }, + { + "cell_type": "code", + "metadata": {}, + "outputs": [], + "execution_count": null, + "source": [ + "%%sql\n", + "SELECT order_id, amount FROM bronze.orders_raw\n" + ] + }, + { + "cell_type": "code", + "metadata": {}, + "outputs": [], + "execution_count": null, + "source": [ + "df = spark.sql(\"SELECT * FROM silver.orders\")\n" + ] + } + ] +} diff --git a/tests/fixtures/re-plan-lake/silver/sales/orders.sql b/tests/fixtures/re-plan-lake/silver/sales/orders.sql new file mode 100644 index 0000000..08a8969 --- /dev/null +++ b/tests/fixtures/re-plan-lake/silver/sales/orders.sql @@ -0,0 +1,2 @@ +CREATE OR REPLACE TABLE silver.orders AS +SELECT order_id, amount FROM bronze.orders_raw; diff --git a/tests/fixtures/re-plan-lake/soda/checks.yml b/tests/fixtures/re-plan-lake/soda/checks.yml new file mode 100644 index 0000000..b310a3c --- /dev/null +++ b/tests/fixtures/re-plan-lake/soda/checks.yml @@ -0,0 +1,3 @@ +checks for orders: + - row_count > 0 + - missing_count(order_id) = 0 diff --git a/tests/fixtures/re-plan-lake/stepfunctions/promote-gold.asl.json b/tests/fixtures/re-plan-lake/stepfunctions/promote-gold.asl.json new file mode 100644 index 0000000..507930f --- /dev/null +++ b/tests/fixtures/re-plan-lake/stepfunctions/promote-gold.asl.json @@ -0,0 +1,7 @@ +{ + "Comment": "Lumenfield gold promotion", + "StartAt": "Promote", + "States": { + "Promote": { "Type": "Task", "Resource": "arn:aws:lambda:us-east-1:123:function:promote", "End": true } + } +} diff --git a/tests/fixtures/re-plan-sql-only/warehouse/dim_customer.sql b/tests/fixtures/re-plan-sql-only/warehouse/dim_customer.sql new file mode 100644 index 0000000..efdae04 --- /dev/null +++ b/tests/fixtures/re-plan-sql-only/warehouse/dim_customer.sql @@ -0,0 +1 @@ +CREATE TABLE dim_customer AS SELECT * FROM staging.customer; diff --git a/tests/test_dekc_plan.py b/tests/test_dekc_plan.py new file mode 100644 index 0000000..174ee38 --- /dev/null +++ b/tests/test_dekc_plan.py @@ -0,0 +1,281 @@ +#!/usr/bin/env python3 +"""Breadth-first RE plan + scoped orchestrate.""" + +from __future__ import annotations + +import json +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +SCRIPTS = ROOT / "scripts" +FIXTURE = ROOT / "tests" / "fixtures" / "re-plan-lake" +SQL_ONLY = ROOT / "tests" / "fixtures" / "re-plan-sql-only" +AUTHOR = "claude-code/lumenfield-detector" + +sys.path.insert(0, str(SCRIPTS)) + +from dekc_common import iter_concept_paths, resolve_author # noqa: E402 +from dekc_plan import ( # noqa: E402 + build_plan, + load_plan, + mark_checklist, + write_plan, +) +from dekc_orchestrate import orchestrate # noqa: E402 +from dekc_walk import extract_notebook_sql # noqa: E402 + + +class TestPlan(unittest.TestCase): + def test_plan_fixture_detects_domains_and_specialists(self): + exports = [ + FIXTURE / "exports" / "fabric-items.json", + FIXTURE / "exports" / "inventory.json", + ] + plan = build_plan([FIXTURE], system_name="Lumenfield Orders", exports=exports) + self.assertEqual(plan["version"], "1") + self.assertEqual(plan["roots"][0]["layout"], "lake") + ecos = set(plan["ecosystems"]) + self.assertTrue({"sql", "dbt", "airflow", "glue", "duckdb", "delta"} <= ecos) + area_ids = {a["id"] for a in plan["focus_areas"]} + for needed in ( + "lake", + "elt", + "orchestration", + "lineage", + "dq", + "bi", + "semantic", + "catalogs", + ): + self.assertIn(needed, area_ids, area_ids) + ranked = [a["id"] for a in sorted(plan["focus_areas"], key=lambda x: x["rank"])] + self.assertEqual(ranked, [a["id"] for a in plan["focus_areas"]]) + agents = {a["id"]: a["agent"] for a in plan["focus_areas"]} + self.assertEqual(agents["lake"], "schema-scout") + self.assertEqual(agents["orchestration"], "stream-job-scout") + self.assertEqual(agents["elt"], "stream-job-scout") + self.assertEqual(agents["lineage"], "lineage-tracer") + self.assertEqual(agents["dq"], "dq-scout") + spec_ids = {s["id"] for s in plan.get("specialists") or []} + for needed in ( + "orch-airflow", + "orch-glue", + "orch-fabric", + "orch-adf", + "orch-stepfunctions", + "orch-composer", + "orch-cron", + "elt-dbt", + "duckdb", + "notebooks", + ): + self.assertIn(needed, spec_ids, spec_ids) + self.assertEqual(agents["orch-airflow"], "airflow-scout") + self.assertEqual(agents["duckdb"], "duckdb-scout") + self.assertEqual(next(a for a in plan["focus_areas"] if a["id"] == "duckdb")["scan_domains"], ["duckdb"]) + self.assertEqual(next(a for a in plan["focus_areas"] if a["id"] == "orch-airflow")["scan_domains"], []) + self.assertTrue(plan.get("cicd_handoff")) + self.assertIn("run-glue.yml", plan["cicd_handoff"][0]["path"]) + for area in plan["focus_areas"]: + self.assertGreaterEqual(area["signal"], 1) + self.assertTrue(area["checklist"]) + for item in area["checklist"]: + self.assertEqual(item["status"], "pending") + + def test_plan_does_not_spawn_absent_specialists(self): + plan = build_plan([SQL_ONLY], system_name="SQL Only") + ids = {a["id"] for a in plan["focus_areas"]} + self.assertIn("lake", ids) + self.assertNotIn("orch-airflow", ids) + self.assertNotIn("orch-glue", ids) + self.assertNotIn("duckdb", ids) + self.assertNotIn("notebooks", ids) + self.assertNotIn("dq", ids) + self.assertEqual((plan.get("specialists") or []), []) + self.assertEqual(plan.get("cicd_handoff") or [], []) + + def test_plan_writes_and_mark_progress(self): + with tempfile.TemporaryDirectory() as td: + bundle = Path(td) / "knowledge" + bundle.mkdir() + written = write_plan(bundle, [FIXTURE], system_name="Lumenfield Orders") + md = bundle / ".dekc" / "re-plan.md" + js = bundle / ".dekc" / "re-plan.json" + self.assertTrue(md.is_file()) + self.assertTrue(js.is_file()) + text = md.read_text(encoding="utf-8") + self.assertIn("# Reverse-engineering plan: Lumenfield Orders", text) + self.assertIn("airflow-scout", text) + self.assertIn("SAC owns", text) + self.assertIn("- [ ] `inventory`", text) + self.assertNotIn("- [x]", text) + marked = mark_checklist(bundle, area="lake", item="inventory", status="done") + self.assertGreaterEqual(marked["checklist"]["done"], 1) + blocked = mark_checklist( + bundle, area="dq", item="capture", status="blocked", note="GE runtime not executed" + ) + self.assertEqual(blocked["checklist"]["blocked"], 1) + reloaded = load_plan(bundle) + lake = next(a for a in reloaded["focus_areas"] if a["id"] == "lake") + inv = next(i for i in lake["checklist"] if i["id"] == "inventory") + self.assertEqual(inv["status"], "done") + text2 = md.read_text(encoding="utf-8") + self.assertIn("- [x] `inventory`", text2) + self.assertIn("blocked", text2) + for p in iter_concept_paths(bundle): + self.assertNotIn(".dekc", p.parts) + + def test_notebook_sql_extract(self): + nb = FIXTURE / "notebooks" / "clean_orders.ipynb" + sqls = extract_notebook_sql(nb) + self.assertGreaterEqual(len(sqls), 2) + blob = "\n".join(sqls).lower() + self.assertIn("select", blob) + self.assertIn("orders", blob) + + +class TestOrchestrate(unittest.TestCase): + def setUp(self): + resolve_author(AUTHOR) + + def test_orchestrate_plan_only(self): + with tempfile.TemporaryDirectory() as td: + host = Path(td) + result = orchestrate( + host, + [FIXTURE], + system_name="Lumenfield Orders", + bundle_name="knowledge", + author=AUTHOR, + exports=[FIXTURE / "exports" / "fabric-items.json"], + plan_only=True, + ) + self.assertEqual(result["phases"], ["init-bundle", "plan"]) + self.assertIsNone(result["walk"]) + areas = {a["id"] for a in result["plan"]["focus_areas"]} + self.assertIn("lake", areas) + self.assertIn("orch-airflow", areas) + spec_ids = {s["id"] for s in result["plan"].get("specialists") or []} + self.assertIn("orch-airflow", spec_ids) + self.assertIn("duckdb", spec_ids) + md = Path(result["bundle"]) / ".dekc" / "re-plan.md" + text = md.read_text(encoding="utf-8") + self.assertIn("- [ ]", text) + self.assertIn("airflow-scout", text) + self.assertIn("re-adversary-judge", text) + for p in iter_concept_paths(Path(result["bundle"])): + self.assertNotIn(".dekc", p.parts) + + def test_orchestrate_from_plan_area_scoped(self): + with tempfile.TemporaryDirectory() as td: + host = Path(td) + planned = orchestrate( + host, + [FIXTURE], + system_name="Lumenfield Orders", + bundle_name="knowledge", + author=AUTHOR, + plan_only=True, + ) + plan_json = Path(planned["bundle"]) / ".dekc" / "re-plan.json" + result = orchestrate( + host, + [FIXTURE], + system_name="Lumenfield Orders", + bundle_name="knowledge", + author=AUTHOR, + from_plan=plan_json, + area="duckdb", + ) + self.assertIn("walk", result["phases"]) + domains = (result["walk"] or {}).get("domains") or [] + self.assertEqual(domains, ["duckdb"]) + sources = list((Path(result["bundle"]) / "sources").glob("*.md")) + sources = [p for p in sources if p.name != "index.md"] + self.assertGreater(len(sources), 0) + text = "\n".join(p.read_text(encoding="utf-8") for p in sources) + self.assertIn("duckdb", text.lower()) + + def test_orchestrate_from_plan_specialist_is_enrichment_only(self): + with tempfile.TemporaryDirectory() as td: + host = Path(td) + planned = orchestrate( + host, + [FIXTURE], + system_name="Lumenfield Orders", + bundle_name="knowledge", + author=AUTHOR, + plan_only=True, + ) + plan_json = Path(planned["bundle"]) / ".dekc" / "re-plan.json" + result = orchestrate( + host, + [FIXTURE], + system_name="Lumenfield Orders", + bundle_name="knowledge", + author=AUTHOR, + from_plan=plan_json, + area="orch-airflow", + ) + self.assertIn("enrichment", result["phases"]) + self.assertNotIn("walk", result["phases"]) + self.assertEqual((result["walk"] or {}).get("domains"), []) + + def test_cli_plan_only_and_mark(self): + with tempfile.TemporaryDirectory() as td: + repo = Path(td) + proc = subprocess.run( + [ + sys.executable, + str(SCRIPTS / "dekc_walk.py"), + str(SQL_ONLY), + "--repo", + str(repo), + "--bundle", + "knowledge", + "--plan-only", + "--system", + "SQL Only", + "--json", + ], + capture_output=True, + text=True, + check=False, + ) + self.assertEqual(proc.returncode, 0, proc.stderr + proc.stdout) + data = json.loads(proc.stdout) + ids = {a["id"] for a in data["focus_areas"]} + self.assertIn("lake", ids) + self.assertNotIn("orch-airflow", ids) + mark = subprocess.run( + [ + sys.executable, + str(SCRIPTS / "dekc_plan.py"), + "mark", + "--repo", + str(repo), + "--bundle", + "knowledge", + "--area", + "lake", + "--item", + "inventory", + "--status", + "done", + "--json", + ], + capture_output=True, + text=True, + check=False, + ) + self.assertEqual(mark.returncode, 0, mark.stderr + mark.stdout) + marked = json.loads(mark.stdout) + self.assertEqual(marked["status"], "done") + + +if __name__ == "__main__": + unittest.main()