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