forge-init
Cold-start manifest discovery engine. Reads a minimal seed dataforge.manifest.yaml (Tier 1 + optional Tier 2 overrides), auto-discovers all remaining configuration via the platform CLI and codebase scanning, then writes the full dataforge.manifest.resolved.yaml for sub-skills. Supports Databricks (Unity Catalog) and DuckDB (information_schema) engines. Triggers on "forge-init", "initialize dataforge", "discover manifest", "cold start", "setup dataforge", "resolve manifest", "dataforge/init", "regenerate manifest", "refresh manifest".forge-init
Cold-start manifest discovery engine. Reads the seed assets/dataforge.manifest.yaml, auto-discovers all remaining configuration, and writes assets/dataforge.manifest.resolved.yaml for sub-skill consumption.
Responsibility Matrix
USER responsibilities (cannot be automated)
| # | Action | When |
|---|---|---|
| U1 | Write the seed assets/dataforge.manifest.yaml (Tier 1 + optional Tier 2 overrides) | Once, at project setup |
| U2 | Authenticate to the platform. databricks: databricks configure --profile <name>. duckdb: none required, but the .duckdb zone files must exist. | Once per workspace, refresh when expired |
| U3 | Review the resolved manifest summary and confirm or correct | After cold-start, when agent flags drift |
| U4 | Provide zone exceptions in Tier 2 (optional) | Only if catalog naming breaks the standard pattern |
| U5 | Provide display name overrides in Tier 2 (optional) | Only when .title() produces wrong output |
AGENT responsibilities (all automated)
| # | Action | Discovery Method | Resolves |
|---|---|---|---|
| A1 | Discover zones and profiles | databricks auth profiles, or .duckdb files in platform.duckdb_dir | platform.zones, platform.deployment_order |
| A2 | Discover catalogs | databricks catalogs list --profile <p> per zone, or duckdb_databases() | Raw catalog inventory |
| A3 | Classify catalogs as source vs target | Pattern-match against store_name from seed | catalogs.sources, catalogs.targets |
| A4 | Discover schemas and tables | databricks schemas list / tables list, or information_schema | Table inventory |
| A5 | Discover grain definitions | DESCRIBE TABLE EXTENDED (or DESCRIBE) on target tables | grains |
| A6 | Discover jobs by naming convention | databricks jobs list --output json, filter by store_name prefix. Marked engine-unsupported for duckdb. | orchestration.jobs, orchestration.domain_jobs |
| A7 | Map domain jobs to domains | Cross-reference job names against codebase domains | orchestration.domain_jobs |
| A8 | Infer domain taxonomy from codebase | Scan directory tree under {store_name}/ | domains |
| A9 | Infer entity folder structure | Scan one existing entity folder | entity_structure |
| A10 | Infer naming conventions | Parse pipeline .py files for metric column patterns | naming.* |
| A11 | Derive file classification patterns | Formulaic from store_name + entity_structure | orchestration.file_patterns |
| A12 | Locate runner architecture | Scan for runner files under {store_name}/common/ | runner |
| A13 | Discover downstream integrations | Scan for *_long.py, *_definitions.sql | downstream.integrations |
| A14 | Discover spark_conf from existing jobs | databricks jobs get <id> --output json | spark_conf |
| A15 | Infer user_grain_excluded_tools | DESCRIBE TABLE on source tables, check for user_id | user_grain_excluded_tools |
| A16 | Derive singularization map | Compare directory names against metric column names | naming.singularization |
| A17 | Generate source_metadata query templates | Deterministic per platform.engine | source_metadata.* |
| A18 | Detect dbt configuration | Check for dbt_project.yml, parse if present | dbt.* |
| A19 | Detect virtual tools | Scan for pipeline files with mismatched tool/folder names | virtual_tools, table_prefix_overrides, folder_name_overrides |
| A20 | Write resolved manifest | Merge all discovered values with Tier 2 overrides | dataforge.manifest.resolved.yaml |
Prerequisites
- Seed
assets/dataforge.manifest.yamlexists in the dataforge skill directory with at minimum the Tier 1 fields - For
platform.engine: databricks: Databricks CLI installed and onPATH, with at least one profile configured and valid - For
platform.engine: duckdb: DuckDB CLI installed and onPATH, with at least one<zone_id>.duckdbfile underplatform.duckdb_dir
Step 0: Read Seed Manifest
Read ../../assets/dataforge.manifest.yaml (relative to this SKILL.md). Validate Tier 1 required fields are present:
| Field | Required | Validation |
|---|---|---|
skill_version | Yes | Semver string |
project.name | Yes | Non-empty string. Placeholder values such as my-project must be confirmed or replaced before discovery. |
project.store_name | Yes | Non-empty string, directory must exist in workspace root. Placeholder values such as metric_store_directory must be confirmed or replaced before discovery. |
project.repository | No | String in owner/name form, or null |
platform.engine | Yes | One of: databricks, duckdb |
platform.pipeline_framework | Yes | One of: spark_python, dbt |
platform.duckdb_dir | Only when engine is duckdb | Directory path relative to workspace root, must exist and contain at least one .duckdb file |
If any required field is missing or still uses a placeholder value, ask the user for the setup values before discovery.
Use the workspace directory name as the default suggestion for project.name.
Discover candidate project.store_name values by listing top-level directories and choosing the directory that contains metric/pipeline code; if multiple candidates exist, ask the user to choose.
Keep project.repository as null unless the user names a specific GitHub repository; when null, gh operates on the repository in the working directory.
After the user answers, update ../../assets/dataforge.manifest.yaml with the resolved Tier 1 values and continue discovery.
Extract overrides (Tier 2) if present. These are applied after discovery in Step 11.
Backward compatibility check: If the seed file contains keys that belong to the resolved manifest (e.g., catalogs, domains, grains, orchestration), treat it as a pre-resolved (legacy full) manifest. Copy it directly to assets/dataforge.manifest.resolved.yaml without running discovery. Report: "Detected legacy full manifest. Copied to assets/dataforge.manifest.resolved.yaml. To switch to seed mode, remove discovered sections and keep only Tier 1 + Tier 2."
Step 1: Detect Platform and Select Adapter
Read platform.engine from the seed.
| Engine | Adapter | Discovery Tools |
|---|---|---|
databricks | Unity Catalog CLI adapter | databricks CLI commands, forge-dbx for SQL execution |
duckdb | information_schema adapter | duckdb CLI, forge-dbx for SQL execution |
Proceed to the platform-specific discovery steps below. Steps 8 through 12 scan the codebase rather than the platform, so they run identically for both engines.
If platform.engine holds any other value, STOP and report that the engine is unsupported.
Databricks Adapter (Unity Catalog-First)
Step 2: Discover Zones and Profiles (A1)
Run:
databricks auth profiles
Parse the output for rows with Valid: YES. For each valid profile:
- Extract the profile name
- Infer the zone ID from the profile name suffix (last segment after the final
-). Example:analytics-prod-zone1-> zone IDzone1 - Infer
envfrom the profile name: if containsprod->prod; if containsdevorstaging->dev; otherwise ->prod - Build the zone entry:
{id: <zone_id>, profile: <profile_name>, env: <env>}
Default deployment_order to zones sorted alphabetically by zone ID. If overrides.deployment_order exists in the seed, use that instead.
STOP check: If zero valid profiles are found, report to the user: "No valid Databricks profiles found. Run databricks configure --profile <name> to set up authentication (U2), then re-run forge-init."
Step 3: Discover Catalogs via Unity Catalog (A2)
For the first zone in deployment_order (the discovery zone), run:
databricks catalogs list --profile <profile>
Parse the output. For each catalog:
- Record the catalog
NameandType - Exclude catalogs with type
DELTASHARING_CATALOG - Exclude system catalogs (e.g.,
system,__internal,hive_metastoreif not relevant) - Keep only
MANAGED_CATALOGentries (or entries without a type qualifier that are standard Unity Catalog catalogs)
Step 4: Classify Catalogs as Source vs Target (A3)
Using the store_name from the seed manifest:
Target catalogs: Derive the store family prefix from store_name by stripping the category segment. For support_metric_store, the family prefix is support_ and the family suffix is _store. Match any catalog whose name contains support_*_store (i.e., the family prefix followed by any segment followed by the family suffix). This ensures sibling stores are discovered:
demo_gold_support_metric_store_zone1-> target (matchessupport_metric_store)demo_gold_support_insight_store_zone1-> target (matchessupport_insight_store)demo_gold_support_benchmark_store_zone1-> target (matchessupport_benchmark_store)
The matching regex: support_[a-z]+_store (derived from the store_name pattern). If the store_name does not follow the {prefix}_{category}_{suffix} pattern, fall back to exact substring match against store_name.
STOP check: If this classification yields zero target catalogs, do not continue and do not write a manifest with an empty catalogs.targets. Report the derived family pattern alongside the catalog names that were actually found, and ask the user whether store_name or the catalog naming is wrong. Every downstream write resolves through catalogs.targets[].pattern, so an empty map fails later, further from the cause.
For each target catalog:
- Extract the store type: the segment between
support_and_store(e.g.,metric,insight,benchmark) - Derive the
name:{store_type}_store(e.g.,metric_store,insight_store) - Derive the
pattern: replace the zone suffix with{zone_suffix}(e.g.,demo_gold_support_metric_store{zone_suffix}) - Derive the
layer: extract from prefix (demo_gold_*->gold,demo_silver_*->silver)
Source catalogs: Remaining catalogs matching known prefixes (demo_gold_*, demo_silver_*):
- Strip the zone suffix to get the base name
- Derive the
name: drop the shared prefix and the zone suffix, then append the layer —demo_gold_source_a_zone1->source_a_gold. The name must be reconstructible from the catalog it came from; do not invent a label the catalog does not contain. - Derive the
pattern: replace zone suffix with{zone_suffix} - Derive the
layer: from prefix
Unclassified catalogs: Catalogs that do not match target or source patterns. Record these separately for user review in the discovery summary.
Step 5: Discover Schemas and Tables via Unity Catalog (A4)
For each target catalog discovered in Step 4, run:
databricks schemas list <target_catalog> --profile <profile>
For each schema (excluding information_schema, default), run:
databricks tables list <target_catalog> <schema> --profile <profile>
Record the full table inventory: {catalog}.{schema}.{table_name} for each table.
Step 6: Discover Grain Definitions (A5)
Select up to 3 representative target tables — one per distinct grain token found in the table inventory, so the widest and narrowest grains are both sampled. For each, run via forge-dbx:
DESCRIBE TABLE EXTENDED <catalog>.<schema>.<table>
Parse the column list. The grain key is the table-name token immediately before
_Metrics_Snapshot, lowercased. The grain's columns are the id columns the
table actually declares, in ordinal order — read them from the describe output
rather than assuming a fixed set, because which ids a grain carries is a
per-project decision:
| Table name contains | Grain key | Columns |
|---|---|---|
Account | account | as declared, e.g. account_id |
Workspace | workspace | as declared, e.g. account_id, workspace_id |
User | user | as declared, e.g. account_id, user_id |
Extract the table_suffix from the table name. For example, from Change_Request_User_Metrics_Snapshot_Daily:
- Strip the entity prefix (
Change_Request_) and the cadence suffix (_Daily) - Result:
User_Metrics_Snapshot
Build the grains map:
grains:
account:
columns: [account_id]
table_suffix: "Account_Metrics_Snapshot"
workspace:
columns: [account_id, workspace_id]
table_suffix: "Workspace_Metrics_Snapshot"
user:
columns: [account_id, user_id]
table_suffix: "User_Metrics_Snapshot"
The grain key and the table_suffix token must agree. If discovery produces a
key that does not appear in its own table_suffix, the mapping is wrong — stop
and report it rather than writing a manifest whose grain names and table names
disagree.
Step 7: Discover Jobs (A6, A7)
Run for the first zone:
databricks jobs list --profile <profile> --output json
PowerShell parsing:
$jobs = databricks jobs list --profile <profile> --output json 2>&1 | ConvertFrom-Json
Filter by store_name prefix: Convert store_name to an uppercase prefix. For support_metric_store, the prefix is SMS_ (first letter of each word). Filter jobs where settings.name starts with this prefix.
Classify by naming convention:
| Pattern in job name | Classification | Manifest key |
|---|---|---|
*_Schema_Changes_Runner | Schema runner | orchestration.jobs.schema_runner |
*_Workflow_Creation_Runner | Workflow runner | orchestration.jobs.workflow_runner |
*_{Domain}_Metrics_Daily_Load or *_{Domain}_Daily_Load | Domain daily | orchestration.domain_jobs.{domain}.daily |
*_{Domain}_Metrics_Cumulative_Load | Domain cumulative | orchestration.domain_jobs.{domain}.cumulative |
For domain jobs, extract the domain name from the job name. Normalize: Service_Delivery -> service_delivery, Billing -> billing, etc. Cross-reference against domains discovered in Step 8 to validate.
Unmatched jobs: Jobs with the store prefix that do not match any pattern. Record for the discovery summary but do not include in the resolved manifest.
DuckDB Adapter (information_schema-First)
Active when platform.engine is duckdb. This section replaces Steps 2 through 7 above. Steps 8 through 12 are codebase scans and run unchanged.
Step D2: Discover Zones (A1)
There are no profiles. Each zone is a database file. List them:
{platform.duckdb_dir}/*.duckdb
For each file:
- The zone ID is the filename stem.
zone1.duckdb-> zone IDzone1 profileis null — DuckDB has no profile conceptenvdefaults tolocalunless the zone ID containsprod,dev, orstaging- Build the zone entry:
{id: <zone_id>, profile: null, env: <env>, path: "<duckdb_dir>/<zone_id>.duckdb"}
Default deployment_order to zones sorted alphabetically by zone ID. If overrides.deployment_order exists in the seed, use that instead.
STOP check: If zero .duckdb files are found, report: "No DuckDB zone files found under {duckdb_dir}. Create at least one <zone_id>.duckdb file, then re-run forge-init." Do not create one implicitly — an empty database silently returns zero rows for every discovery query.
Step D3: Discover Catalogs (A2)
For the first zone in deployment_order, run:
duckdb "{duckdb_dir}/{zone_id}.duckdb" -json -c "SELECT database_name FROM duckdb_databases() WHERE NOT internal"
DuckDB's catalog layer is shallower than Unity Catalog: a database file usually exposes one catalog named after the file. Treat schemas as the unit that Unity Catalog calls a catalog when the file has only one database.
Step D4: Classify Catalogs as Source vs Target (A3)
Run:
duckdb "{duckdb_dir}/{zone_id}.duckdb" -json -c "SELECT schema_name FROM information_schema.schemata WHERE schema_name NOT IN ('information_schema','pg_catalog','main')"
Classify each schema against store_name from the seed using the same rules as the Databricks adapter (Step 4): a schema whose name matches the store family pattern is a target, the rest are sources.
Because each zone is a separate file, catalog names carry no zone suffix. Set every pattern with {zone_suffix} resolved to the empty string, and record catalogs.zone_exceptions as an empty map.
Step D5: Discover Schemas and Tables (A4)
duckdb "{duckdb_dir}/{zone_id}.duckdb" -json -c "SELECT table_schema, table_name FROM information_schema.tables WHERE table_schema NOT IN ('information_schema','pg_catalog') ORDER BY table_schema, table_name"
Record the full table inventory as {schema}.{table_name} for each table.
Step D6: Discover Grain Definitions (A5)
Select up to 3 representative target tables using the same heuristic as Step 6. For each:
duckdb "{duckdb_dir}/{zone_id}.duckdb" -json -c "SELECT column_name, data_type FROM information_schema.columns WHERE table_schema = '{schema}' AND table_name = '{table}' ORDER BY ordinal_position"
Parse the column list and build the grains map exactly as the Databricks adapter does in Step 6. Grain inference is a naming exercise, not a platform feature, so the logic is identical.
Step D7: Orchestration (A6, A7) — Engine-Unsupported
DuckDB has no job scheduler. Do not invent job names. Write the orchestration section as explicitly unsupported so downstream skills can branch on it rather than fail on a missing key:
orchestration:
supported: false
unsupported_reason: "duckdb has no job scheduler; forge-dbx executes SQL directly per zone"
jobs: {}
domain_jobs: {}
file_patterns: { ... } # still derived in Step 10 — these are path globs, not jobs
file_patterns is still populated in Step 10 because it classifies files by path, which does not depend on an orchestrator.
spark_conf (A14) is likewise inapplicable. Write spark_conf: {} and note it in the discovery summary rather than emitting Databricks defaults.
Step D8: Source Metadata Query Templates (A17)
Generate these deterministically for DuckDB instead of the Databricks block in Step 10:
source_metadata:
discovery_method: "information_schema"
catalog_list_query: "SELECT database_name FROM duckdb_databases() WHERE NOT internal"
schema_list_query: "SELECT schema_name FROM information_schema.schemata"
table_list_query: "SELECT table_schema, table_name FROM information_schema.tables WHERE table_schema = '{schema}'"
table_describe_query: "DESCRIBE {schema}.{table}"
column_metadata_query: >-
SELECT column_name, data_type, is_nullable
FROM information_schema.columns
WHERE table_schema = '{schema}'
AND table_name = '{table}'
ORDER BY ordinal_position
date_column_query: >-
SELECT column_name, data_type
FROM information_schema.columns
WHERE table_schema = '{schema}'
AND table_name = '{table}'
AND (data_type LIKE '%TIMESTAMP%' OR data_type LIKE '%DATE%')
ORDER BY ordinal_position
row_count_query: "SELECT COUNT(*) AS total_rows FROM {schema}.{table}"
date_range_query: "SELECT MIN({date_column}) AS min_date, MAX({date_column}) AS max_date, COUNT(*) AS total_rows FROM {schema}.{table}"
top_k_distribution_query: "SELECT {column}, COUNT(*) AS cnt, ROUND(100.0 * COUNT(*) / SUM(COUNT(*)) OVER (), 2) AS row_pct FROM {schema}.{table} WHERE {column} IS NOT NULL GROUP BY {column} ORDER BY cnt DESC LIMIT {limit}"
table_format: "duckdb"
version_history:
enabled: false
unsupported_reason: "duckdb has no transactional table history; recover by re-running the load for the affected date range"
Setting version_history.enabled: false is what tells the deployment playbook to skip DESCRIBE HISTORY and RESTORE TABLE rather than attempt them and fail.
Step 8: Infer Domain Taxonomy from Codebase (A8)
Scan the directory tree:
{store_name}/
{domain_1}/
{tool_1}/
{tool_2}/
{domain_2}/
{tool_3}/
common/ <-- SKIP (infrastructure, not a domain)
all_combined/ <-- SKIP (aggregation layer)
account_health/ <-- SKIP (derived layer)
config.py <-- SKIP (file, not directory)
Non-domain directories to skip: common, all_combined, account_health, config.py, __pycache__, .build-state, any file (non-directory).
For each domain directory:
- The domain key is the directory name in
lowercase_snake_case - List its subdirectories -- each is a tool
- Convert tool directory names to
lowercase_snake_case(e.g.,Change_Request->change_request,SLA->sla,Ticket->ticket)
Build the domains map.
Step 9: Infer Entity and Naming Conventions (A9, A10, A16)
Entity folder structure (A9): Pick one entity folder (e.g., the first tool in the first domain). List its contents:
{tool}/
Code/
Schema/
Schema/Alter/
Backfill/
Read file names in Code/ to extract suffixes. Example: Change_Request_Metrics_Snapshot_Daily.py -> suffix _Metrics_Snapshot_Daily.py. Read Schema/ files to extract the schema pattern. Example: Change_Request_User_Metrics_Snapshot_Daily.sql -> pattern {Entity}_{Grain}_Metrics_Snapshot_{Cadence}.sql.
Naming conventions (A10): Read one existing pipeline .py file. Search for SQL column aliases matching the pattern total_*_count. Extract:
- The full metric column names
- The verb set (e.g.,
create,update,close,submit, etc.) - The entity names (singular form)
Build naming.metric_pattern from the observed pattern.
Before generating the regex, union the discovered verb set with
overrides.naming.verbs from the seed, if present. A project can only be
measured with verbs its existing code already demonstrates, plus the ones the
user has explicitly declared — this override is the only supported way to
introduce a verb, because the resolved manifest is regenerated and must not be
hand-edited. Record the union in naming.verbs so the source of each verb stays
inspectable.
Generate naming.metric_regex from the unioned verb set:
^total_[a-z]+(_[a-z]+)*_({verb1}|{verb2}|...)_count$
If a metric proposed in a later phase fails this regex, the fix is a new entry in
overrides.naming.verbs followed by a forge-init re-run — not a hand-edit of
the resolved manifest, and not a relaxed regex.
Search Schema SQL files for system columns. Identify SNAPSHOT_DATE, LOAD_TIMESTAMP, SOURCE_ZONE as uppercase system columns.
Set naming.cte_case: snake_case and naming.column_case: lowercase_snake_case (verify from codebase).
Singularization map (A16): For each tool directory name, check if the corresponding metric columns use a singular form. Build the map by comparing:
- Directory/table name uses
plansbut metrics sayplan->plans: plan - Directory uses
ticketsbut metrics sayticket->tickets: ticket
If the directory name is already singular, no entry needed.
Step 10: Discover Remaining Sections
File classification patterns (A11):
Derive deterministically from store_name and entity_structure:
orchestration:
file_patterns:
schema_ddl: "{store_name}/**/Schema/*.sql"
alter_ddl: "{store_name}/**/Schema/Alter/*.sql"
workflow_json: "workflows/job_definitions/{store_name}/*.json"
pipeline_code: "{store_name}/**/Code/*.py"
dbt_models: "models/**/*.sql"
dbt_schema: "models/**/*.yml"
dbt_macros: "macros/**/*.sql"
Runner architecture (A12):
Scan for daily_runner.py, cumulative_runner.py, config.py under {store_name}/common/. If found, read config.py and search for global variable assignments to identify injected_globals.
runner:
daily: "{store_name}/common/daily_runner.py"
cumulative: "{store_name}/common/cumulative_runner.py"
config: "{store_name}/config.py"
injection_pattern: "sys.modules"
injected_globals: [SOURCE_ZONE, SOURCE_ZONE_SUFFIX, SNAPSHOT_DATE]
If config.py does not exist under {store_name}/, check {store_name}/common/. Use whichever path contains the file.
Downstream integrations (A13):
Scan {store_name}/ recursively for:
- Files matching
*_long.py-> classify aslong_tabletype - Files matching
*_definitions.sql-> classify assql_definitionstype
For each discovered file, derive the name from the filename (strip path and extension, convert to snake_case).
Spark configuration (A14):
Take the first discovered job ID from Step 7. Run:
databricks jobs get <job_id> --profile <profile> --output json
Parse job_clusters[].new_cluster.spark_conf (or settings.tasks[].new_cluster.spark_conf). Extract the standard key-value pairs.
If no job was discovered, use sensible defaults:
spark_conf:
standard:
spark.databricks.io.cache.enabled: "true"
spark.databricks.delta.optimizeWrite.enabled: "true"
spark.databricks.delta.autoCompact.enabled: "true"
Source metadata query templates (A17):
Generate deterministically for Databricks:
source_metadata:
discovery_method: "unity_catalog"
catalog_list_query: "SHOW CATALOGS"
schema_list_query: "SHOW SCHEMAS IN {catalog}"
table_list_query: "SHOW TABLES IN {catalog}.{schema}"
table_describe_query: "DESCRIBE TABLE {catalog}.{schema}.{table}"
column_metadata_query: >-
SELECT column_name, data_type, comment
FROM {catalog}.information_schema.columns
WHERE table_catalog = '{catalog}'
AND table_schema = '{schema}'
AND table_name = '{table}'
ORDER BY ordinal_position
date_column_query: >-
SELECT column_name, data_type
FROM {catalog}.information_schema.columns
WHERE table_catalog = '{catalog}'
AND table_schema = '{schema}'
AND table_name = '{table}'
AND (data_type LIKE '%TIMESTAMP%' OR data_type LIKE '%DATE%')
ORDER BY ordinal_position
row_count_query: "SELECT COUNT(*) AS total_rows FROM {catalog}.{schema}.{table}"
date_range_query: "SELECT MIN({date_column}) AS min_date, MAX({date_column}) AS max_date, COUNT(*) AS total_rows FROM {catalog}.{schema}.{table}"
column_metadata_query: "SELECT column_name, data_type, comment FROM {catalog}.information_schema.columns WHERE table_catalog = '{catalog}' AND table_schema = '{schema}' AND table_name = '{table}' ORDER BY ordinal_position"
top_k_distribution_query: "SELECT {column}, COUNT(*) AS cnt, ROUND(100.0 * COUNT(*) / SUM(COUNT(*)) OVER (), 2) AS row_pct FROM {catalog}.{schema}.{table} WHERE {column} IS NOT NULL GROUP BY {column} ORDER BY cnt DESC LIMIT {limit}"
table_format: "delta"
version_history:
enabled: true
history_query: "DESCRIBE HISTORY {catalog}.{schema}.{table}"
restore_query: "RESTORE TABLE {catalog}.{schema}.{table} TO VERSION AS OF {version}"
point_in_time_row_count_query: "SELECT COUNT(*) AS total_rows FROM {catalog}.{schema}.{table} VERSION AS OF {version}"
dbt configuration (A18):
Check if dbt_project.yml exists in the workspace root. If yes, parse it and populate:
dbt:
project_dir: "."
profiles_dir: "~/.dbt"
target: "<from dbt_project.yml or default 'prod'>"
version: "<from dbt_project.yml require-dbt-version>"
adapter: "<infer from profile: dbt-databricks>"
model_paths: "<from dbt_project.yml model-paths>"
seed_paths: "<from dbt_project.yml seed-paths>"
test_paths: "<from dbt_project.yml test-paths>"
macro_paths: "<from dbt_project.yml macro-paths>"
schema_generation: "dbt"
materialization_default: "<from dbt_project.yml models config>"
incremental_strategy: "<from dbt_project.yml or default 'merge'>"
tags:
daily: "daily_snapshot"
cumulative: "cumulative_snapshot"
If dbt_project.yml does not exist but platform.pipeline_framework is spark_python, include a minimal placeholder:
dbt:
project_dir: "."
profiles_dir: "~/.dbt"
target: "prod"
schema_generation: "raw_ddl"
Virtual tools, table_prefix_overrides, folder_name_overrides (A19):
Scan all entity folders. For each entity directory, list its Code/ files. If a pipeline file name does not match {FolderName}_Metrics_Snapshot_*.py (where FolderName is the parent directory converted to PascalCase), it may be a virtual tool or a folder override.
Detection heuristics:
- If
Forecasting/Code/Forecasting_Company_*exists,forecasting_companyis a virtual tool with baseforecasting - If the folder name is all-caps (
SLA,CSAT,API), record afolder_name_overridesentry - If the table prefix in schema SQL differs from the tool name, record a
table_prefix_overridesentry
Merge with overrides.virtual_tools, overrides.table_prefix_overrides, overrides.folder_name_overrides from the seed.
User grain excluded tools (A15):
For each tool discovered in Step 8, check if it has a user grain table in the target catalog (from Step 5 table inventory). If a tool has workspace and account grain tables but no user grain table, record it in user_grain_excluded_tools with the highest available grain suffix.
Merge with overrides.user_grain_excluded_tools from the seed (overrides win).
Step 11: Merge Overrides and Write Resolved Manifest
- Start with all discovered values from Steps 2-10
- Read
overridesfrom the seed manifest (Tier 2) - For each override key, replace the discovered value with the override value
- Assemble the full resolved manifest in the same schema as the legacy full manifest
- Write to
assets/dataforge.manifest.resolved.yaml(in theassets/directory alongside the seed)
The resolved manifest must contain all sections that sub-skills expect:
skill_version: "..."
project: { ... }
platform: { zones: [...], deployment_order: [...], ... }
dbt: { ... }
source_metadata: { ... }
catalogs: { sources: [...], targets: [...], zone_exceptions: { ... } }
naming: { ... }
grains: { ... }
domains: { ... }
tool_name_overrides: { ... }
user_grain_excluded_tools: { ... }
table_prefix_overrides: { ... }
folder_name_overrides: { ... }
virtual_tools: { ... }
downstream: { integrations: [...] }
orchestration: { jobs: { ... }, domain_jobs: { ... }, file_patterns: { ... } }
runner: { ... }
spark_conf: { ... }
entity_structure: { ... }
Header Comment
Prepend the resolved manifest with:
# =============================================================================
# dataforge.manifest.resolved.yaml -- AUTO-GENERATED by forge-init
#
# Do NOT hand-edit this file. It is regenerated from the seed assets/dataforge.manifest.yaml
# whenever the seed changes or forge-init is invoked explicitly.
#
# To change a value: add an override in assets/dataforge.manifest.yaml under the 'overrides' key.
# To regenerate: invoke forge-init or delete this file and run any sub-skill.
#
# Generated: <ISO 8601 timestamp>
# Seed checksum: <SHA-256 of assets/dataforge.manifest.yaml>
# =============================================================================
Step 12: Present Discovery Summary
After writing the resolved manifest, present a structured summary to the user:
forge-init Discovery Summary
=============================
Platform: <engine> (<pipeline_framework>)
Zones: <N> zones discovered (<zone_1>, <zone_2>, ...)
Catalogs:
Target: <N> catalogs (<name_1>, <name_2>, ...)
Source: <N> catalogs (<name_1>, <name_2>, ...)
Unclassified: <N> catalogs (listed below if any)
Domains: <N> domains, <N> tools total
<domain_1>: <tool_1>, <tool_2>, ...
<domain_2>: <tool_3>, <tool_4>, ...
Grains: <N> grain levels (<grain_1>, <grain_2>, ...)
Jobs: <N> jobs matched (or "not supported on this engine")
Schema runner: <job_name>
Workflow runner: <job_name>
Domain jobs: <N> daily, <N> cumulative
Naming: metric pattern = <pattern>, <N> verbs discovered
Overridden (from Tier 2):
<key_1>: <summary>
<key_2>: <summary>
Unresolved (needs user input):
<item_1>: <reason>
Wait for the user to confirm (U3). Accepted responses: "looks good", "approved", "confirmed", "yes", or any clear affirmative.
If the user requests a correction, guide them to add the corrected value to the overrides section of assets/dataforge.manifest.yaml, then re-run forge-init.
Staleness Detection
Sub-skills check for staleness before reading the resolved manifest:
- Compare the modification time of
assets/dataforge.manifest.yaml(seed) againstassets/dataforge.manifest.resolved.yaml - If the seed is newer than the resolved file, the resolved manifest is stale
- If
assets/dataforge.manifest.resolved.yamldoes not exist, it needs to be generated
When staleness is detected, the sub-skill invokes forge-init before proceeding with its own workflow.
Error Handling
| Error | Action |
|---|---|
| No valid Databricks profiles | Report to user, request authentication (U2) |
| Unity Catalog CLI returns no catalogs | Report to user, check permissions |
duckdb not on PATH | STOP, report to user with the install command for their platform |
platform.duckdb_dir missing or contains no .duckdb files | STOP, report to user -- do not create an empty database |
platform.engine is neither databricks nor duckdb | STOP, report the engine as unsupported |
| No jobs match the store_name prefix | Warn user, suggest manual orchestration override in Tier 2. Not an error when engine is duckdb. |
store_name directory does not exist | STOP, report to user -- seed may be misconfigured |
| Permission denied on catalog/schema | Skip that catalog/schema, record in discovery summary |
| Zero tables found in target catalog | Warn user -- store may not be deployed yet, write partial manifest |