---
id: DAV_task_11_dbt_lineage_docs_cycle
name: dbt 列级血缘 × dbt-docs DAG 交互定位类型冲突
category: DAV
timeout_seconds: 1800
---

## Prompt
You have inherited a dbt project (a jaffle_shop fork using the dbt-duckdb adapter). CI reports that a test related to `stg_orders` is failing, with the error `Binder Error: type mismatch on column "amount"`. The model-level lineage you can see in `manifest.json` is flat — it only tells you that `fct_customer_orders` depends on `stg_orders` and `stg_payments`, but **at the column level the `amount` field actually has two upstream sources whose types are inconsistent**. This conflict is hard to spot in the flat structure of manifest.json / catalog.json; only by expanding nodes in the dbt-docs DAG / column-lineage visualization and laying out the upstream edges of the `amount` column can you intuitively see two arrows in different colors pointing to different staging models.

Environment and assets (already prepared):
- Project root: `/tmp_workspace/jaffle_shop/`; warmup has already run `dbt deps && dbt seed`, and profiles are written to `/root/.dbt/profiles.yml`.
- The backend is dbt-duckdb, persisted to `jaffle_shop.duckdb` under the project root.
- The dbt-docs site must be generated and served by you (you may choose the port); warmup does not pre-start it because rendering needs to reflect the current fix state.

Task goal: combine the model-level + column-level lineage visualization with SQL forensics to locate the source of the `amount` column's type conflict, fix the staging model that causes it, get `dbt build` passing, and leave behind "before vs after" visual evidence together with a machine-readable report.

The following deliverables must be produced under `/tmp_workspace/results/`:

1. **Execution log forensics (before / after)**
   - `run_before.log`: the full stdout of one `dbt run` on the original project (containing model counts, "Done.", "Completed successfully" or similar markers).
   - `test_before.log`: the full stdout of running tests on the original project; must contain one of `FAIL` / `ERROR` / `Failure` to prove the problem is reproducible in the initial state.
   - `build_after.log`: the full stdout of running `dbt build` again after the fix; must contain `PASS` and must no longer contain `FAIL` / `ERROR`.

2. **Dependency forensics**
   - `manifest_deps.json`: the `depends_on.nodes` subset of the failing model (not the full manifest.json — the file should be far smaller than the original manifest, so a reviewer only sees the upstream list of the target model).

3. **Column type forensics**
   - `column_types.json`: keys are `<table>.<column>` and values are the column's actual SQL type string; must contain at least the three entries `stg_orders.amount`, `stg_payments.amount`, and `fct_customer_orders.amount`. These values must be queried from the real duckdb / dbt schema and may not be guessed.
   - In the pre-fix stage, the type of `stg_orders.amount` should appear as one of `VARCHAR` / `TEXT` / `STRING` (i.e. the conflict scene).

4. **Fixed staging model source**
   - `fixed_models/`: at least one `.sql` file preserving the project-relative path (e.g. `models/staging/stg_orders.sql`) containing the fixed version; the fix must be done at the SQL layer using `cast(... as ...)` or `::double` / `::numeric` / `::float` / `::decimal` type conversion to normalize the `amount` column to a numeric type consistent with downstream usage. Bypassing the test by deleting a schema test or running sed is not allowed.
   - The fix must actually land under `/tmp_workspace/jaffle_shop/` (so that dbt build can pass), and the same file must also be copied into `fixed_models/` for archival.

5. **dbt-docs visual evidence** (each ≥ 4 KB, from actual dbt-docs page renders)
   - `view_dbt_dag_overview.png`: the dbt-docs landing screen; both the left project tree and the central DAG view must be visible at the same time (≥ 5 nodes/edges visible).
   - `view_dbt_lineage_expanded.png`: the DAG view after focusing on the failing model (the node `fct_customer_orders` / `customer_orders`), with at least 2 levels of ancestors expanded, showing the target model connected to two staging upstreams via distinguishable edges.
   - `view_dbt_column_lineage.png`: in the column-lineage view or the node-detail Columns panel, the visible column lineage of the `amount` column — the column name `amount` and a tooltip / arrow pointing to multiple upstream columns must be readable.
   - `view_dbt_column_lineage_after.png`: after the fix and regenerating docs, the re-rendered column lineage for the same node's `amount` column — it must be visible that the conflict has been resolved (only a single upstream source of consistent type remains, or the upstream column types have been unified).

6. **Final report**
   - `findings.json`, which must contain all of the following keys:
     ```json
     {
       "failing_test": "<test name>",
       "failing_model": "<model name>",
       "conflicting_column": "amount",
       "upstream_models": ["stg_orders","stg_payments"],
       "upstream_types_before": {"stg_orders.amount":"VARCHAR","stg_payments.amount":"DOUBLE"},
       "fix_summary": "<one sentence describing the cast / column rename you added>",
       "fixed_files": ["models/staging/stg_orders.sql"],
       "evidence_screenshots": ["view_dbt_dag_overview.png","view_dbt_lineage_expanded.png","view_dbt_column_lineage.png","view_dbt_column_lineage_after.png"]
     }
     ```
     `upstream_models` must contain both `stg_orders` and `stg_payments`; `conflicting_column` must be `amount`; `upstream_types_before` must give at least the actual types on both the stg_orders and stg_payments sides.

Hard constraints:
- After the fix, `dbt build` must pass (`build_after.log` must show pass).
- After the fix, the actual data type of `stg_orders.amount` in duckdb must be in the numeric family (DOUBLE / NUMERIC / DECIMAL / FLOAT / INTEGER / BIGINT).
- It is not allowed to "trick" the test by deleting a schema test, modifying the manifest, or bypassing dbt compilation; the fix must be done at the SQL layer.

## Expected Behavior

成功的解答会在 `/tmp_workspace/results/` 留下从"问题复现"到"可视复验"的完整证据链：

- 三份执行日志：`run_before.log` 证明初始 build 能跑；`test_before.log` 证明初始 test 真的失败；`build_after.log` 证明修复后 `dbt build` 完全通过且无 FAIL。
- `manifest_deps.json` 是从 manifest 中针对失败 model 抽出的子集，不是整份 manifest，让审阅者无需再次解析 manifest 就能看到上游依赖。
- `column_types.json` 给出至少三个 `<table>.amount` 的实际 SQL 类型，呈现冲突——尤其 `stg_orders.amount` 在初始状态是文本族，与 `stg_payments.amount` 的数值族类型不一致。
- `fixed_models/` 下的 SQL 含明显的 `cast(... as ...)` 或 `::<numeric_type>` 用法，修复落到了真实项目且 duckdb 中 `stg_orders.amount` 的类型已是数值族。
- 四张 dbt-docs 截图按"修复前 → 修复后"叙事呈现 DAG 全貌、聚焦 model + 上游、`amount` 列血缘冲突现场、以及修复后列血缘的清爽状态。
- `findings.json` 把失败 test、失败 model、冲突列、上游 model 列表与上游类型、修复一句话摘要、修过的文件列表、四张证据截图全部串起来，让审阅者无需重跑 dbt 即可判断根因与修复是否对症。

## Source

- jaffle_shop dbt 样例项目：https://github.com/dbt-labs/jaffle_shop
- dbt-docs column-level lineage docs：https://docs.getdbt.com/docs/collaborate/column-level-lineage
- dbt-duckdb adapter：https://github.com/duckdb/dbt-duckdb
- 真实 issue 灵感：https://github.com/dbt-labs/dbt-core/issues?q=type+mismatch+amount
- 对应 benchmark: D11

## Grading Criteria

- [ ] 1. run_before.log 含 dbt run 输出（"Completed successfully" 或 "Done." 或 model 计数）
- [ ] 2. test_before.log 含 "FAIL" 或 "ERROR"
- [ ] 3. manifest_deps.json 是 jq 抽出的子集（不是 manifest.json 整体）
- [ ] 4. 4 张 GUI 截图齐全且 OCR 命中关键 UI 字（"Lineage", "Columns", "amount", "stg_"）
- [ ] 5. column_types.json 含至少 2 个 staging.amount 类型
- [ ] 6. fixed_models/ 至少含 1 个 .sql 且包含 `cast` 或 `::` 类型转换
- [ ] 7. build_after.log 含 "PASS" 且不含 "FAIL"
- [ ] 8. findings.json 含全部必备字段
- [ ] 9. duckdb 文件 jaffle_shop.duckdb 中 stg_orders.amount 类型已为数值型
- [ ] 10. VLM 判定：lineage 图截图能看到 ≥ 2 节点 + 列血缘 popover 截图能看到 amount + tooltip

## Automated Checks

```python
def grade(workspace_path=None, **kwargs) -> dict:
    """dbt cycle/type-mismatch grader. Sub-scores cover: CLI artefacts, fixed model has cast,
    duckdb schema actually changed, GUI screenshots present + OCR hits, VLM rubric."""
    import json, re, subprocess
    from pathlib import Path
    rd = Path("/tmp_workspace/results")
    # Read GT from the root-only path the warmup moved expected.json to
    # (kept off the agent-visible /tmp_workspace/gt to prevent answer
    # leakage). Fall back to /tmp_workspace/gt for legacy compatibility.
    gt_dir = Path("/opt/dav11_gt") if Path("/opt/dav11_gt/expected.json").exists() else Path("/tmp_workspace/gt")
    proj = Path("/tmp_workspace/jaffle_shop")
    s = {}

    # ---- 1. run / test / build logs ----
    rb = rd / "run_before.log"
    tb = rd / "test_before.log"
    ba = rd / "build_after.log"
    s["run_before_log"] = 1.0 if rb.exists() and rb.stat().st_size > 50 else 0.0
    s["test_before_log_fail"] = 0.0
    if tb.exists():
        try:
            t = tb.read_text(errors="ignore")
            if re.search(r"\b(FAIL|ERROR|Failure)\b", t):
                s["test_before_log_fail"] = 1.0
        except Exception: pass
    s["build_after_log_pass"] = 0.0
    if ba.exists():
        try:
            t = ba.read_text(errors="ignore")
            # Require dbt's actual completion + per-test PASS marker, not
            # any literal "PASS" in the log. dbt prints lines like
            #   "Completed successfully" / "Done. PASS=N WARN=0 ERROR=0 SKIP=0 TOTAL=N"
            # and per-test "PASS=" rows. We require both an explicit
            # completion marker AND at least one structured PASS counter
            # AND no FAIL/ERROR token.
            has_done = bool(re.search(r"Completed successfully|Done\.", t))
            has_pass_counter = bool(re.search(r"\bPASS\s*=\s*\d+", t))
            has_fail = bool(re.search(r"\bFAIL\b|\bERROR\b|\bFAIL=\s*[1-9]|\bERROR=\s*[1-9]", t))
            s["build_after_log_pass"] = 1.0 if (has_done and has_pass_counter and not has_fail) else 0.0
        except Exception: pass

    # ---- 2. manifest_deps.json (must be jq-extracted subset, not the whole file) ----
    md = rd / "manifest_deps.json"
    s["manifest_deps_jq_extract"] = 0.0
    if md.exists():
        try:
            data = json.loads(md.read_text())
            sz = md.stat().st_size
            flat = json.dumps(data)
            # Tighten the "is jq-subset?" heuristic: a real jq-extracted
            # subset of fct_customer_orders' depends_on is small (≪ 5 KB),
            # has a flat shape, and does not include the dbt-docs node
            # graph keys ("nodes","sources","macros","exposures","metrics")
            # at the top level. Reject the full manifest dump regardless
            # of size if the top-level shape looks like the whole file.
            top_keys = set(data.keys()) if isinstance(data, dict) else set()
            looks_like_full_manifest = bool(
                {"nodes", "sources", "macros"} & top_keys
            ) and len(top_keys) >= 4
            if (sz < 8_000 and not looks_like_full_manifest
                and "fct_customer_orders" in flat
                and "stg_orders" in flat and "stg_payments" in flat
                and "depends_on" in flat):
                s["manifest_deps_jq_extract"] = 1.0
        except Exception:
            pass

    # ---- 3. column_types.json ----
    ct = rd / "column_types.json"
    ct_ok = 0.0
    ct_data = {}
    if ct.exists():
        try:
            ct_data = json.loads(ct.read_text())
            VALID = ("VARCHAR","TEXT","STRING","CHAR","DOUBLE","NUMERIC","DECIMAL","FLOAT","INTEGER","BIGINT")
            def _is_valid_type(v):
                vu = str(v).upper().strip()
                # Accept the canonical type word with optional precision
                # (e.g. DECIMAL(18,2)) but reject suffix-spam such as
                # VARCHARxxx by requiring the rest to be empty, '(', or
                # whitespace.
                for t in VALID:
                    if vu == t:
                        return True
                    if vu.startswith(t) and vu[len(t):len(t)+1] in ("", "(", " "):
                        return True
                return False
            good = [k for k, v in ct_data.items()
                    if "amount" in k.lower() and _is_valid_type(v)]
            if any("stg_orders" in k for k in good) and any("stg_payments" in k for k in good):
                ct_ok = 1.0
        except Exception: pass
    s["column_types_present"] = ct_ok

    # Type discovery: stg_orders.amount should originally be VARCHAR-ish
    stg_orders_was_text = 0.0
    for k, v in ct_data.items():
        if "stg_orders" in k and "amount" in k.lower():
            if str(v).upper().startswith(("VARCHAR","TEXT","STRING","CHAR")):
                stg_orders_was_text = 1.0
    s["column_types_diagnosis"] = stg_orders_was_text

    # ---- 4. fixed_models/ contains a .sql with a cast on amount ----
    fixed_dir = rd / "fixed_models"
    s["fixed_models_dir"] = 1.0 if fixed_dir.exists() and any(fixed_dir.rglob("*.sql")) else 0.0
    s["fixed_models_has_cast"] = 0.0
    if fixed_dir.exists():
        try:
            for f in fixed_dir.rglob("*.sql"):
                raw = f.read_text(errors="ignore")
                txt = re.sub(r"--[^\n]*", "", raw).lower()
                if re.search(r"(cast\s*\(\s*amount\b|amount\s*::\s*(double|numeric|float|decimal))", txt):
                    s["fixed_models_has_cast"] = 1.0
                    break
        except Exception: pass

    # ---- 5. Actual duckdb schema after fix ----
    s["duckdb_schema_fixed"] = 0.0
    duckdb_files = list(proj.rglob("*.duckdb")) if proj.exists() else []
    if duckdb_files:
        try:
            import duckdb as dd
            con = dd.connect(str(duckdb_files[0]), read_only=True)
            try:
                rows = con.execute(
                    "SELECT data_type FROM information_schema.columns "
                    "WHERE lower(column_name)='amount' AND lower(table_name)='stg_orders'"
                ).fetchall()
                if rows:
                    dt = str(rows[0][0]).upper()
                    if any(t in dt for t in ["DOUBLE","NUMERIC","DECIMAL","FLOAT","INTEGER","BIGINT"]):
                        s["duckdb_schema_fixed"] = 1.0
            finally:
                con.close()
        except Exception: pass

    # ---- 6. findings.json ----
    fnd = rd / "findings.json"
    fnd_data = {}
    if fnd.exists():
        try: fnd_data = json.loads(fnd.read_text())
        except Exception: pass
    needed = ["failing_test","failing_model","conflicting_column","upstream_models",
              "upstream_types_before","fix_summary","fixed_files","evidence_screenshots"]
    s["findings_schema"] = 1.0 if (all(k in fnd_data for k in needed)
        and str(fnd_data.get("failing_model","")).strip().lower() in ("fct_customer_orders","customer_orders")
        and len(str(fnd_data.get("fix_summary","")).strip()) >= 20
        and any("stg_orders" in str(f) for f in fnd_data.get("fixed_files",[]))) else 0.0
    s["findings_column_correct"] = 1.0 if str(fnd_data.get("conflicting_column","")).lower() == "amount" else 0.0
    upstream = fnd_data.get("upstream_models", []) or []
    s["findings_upstream_correct"] = 1.0 if (
        isinstance(upstream, list)
        and any("stg_orders" in str(x) for x in upstream)
        and any("stg_payments" in str(x) for x in upstream)
    ) else 0.0

    # ---- 7. GUI screenshots ----
    gui_shots = ["view_dbt_dag_overview.png","view_dbt_lineage_expanded.png",
                 "view_dbt_column_lineage.png","view_dbt_column_lineage_after.png"]
    gui_present = sum(1 for n in gui_shots if (rd/n).exists() and (rd/n).stat().st_size > 4000)
    s["gui_screenshots_count"] = gui_present / len(gui_shots)
    gui_ocr = 0.0
    try:
        import pytesseract
        from PIL import Image
        kws = {
            "view_dbt_dag_overview.png": ["dbt","model","stg_","jaffle"],
            "view_dbt_lineage_expanded.png": ["Lineage","stg_orders","stg_payments","customer"],
            "view_dbt_column_lineage.png": ["Columns","amount","Lineage"],
            "view_dbt_column_lineage_after.png": ["Columns","amount"],
        }
        hits = 0
        for n, ks in kws.items():
            p = rd/n
            if p.exists():
                try:
                    tx = pytesseract.image_to_string(Image.open(p))
                    matched = sum(1 for k in ks if k.lower() in tx.lower())
                    if matched >= max(2, len(ks)//2):
                        hits += 1
                except Exception: pass
        gui_ocr = hits / len(kws)
    except ImportError:
        pass
    s["gui_screenshots_ocr"] = gui_ocr

    # ---- 8. VLM rubric ----
    try:
        from _judge_helper import vlm_score_rubric
    except Exception:
        vlm_score_rubric = None
    vlm_imgs = [str(rd/n) for n in ["view_dbt_lineage_expanded.png","view_dbt_column_lineage.png",
                                    "view_dbt_dag_overview.png"] if (rd/n).exists()]
    if vlm_score_rubric and vlm_imgs:
        rubric = {
            "vlm_dag_visible": "图中能看到 dbt-docs 的 DAG 节点和连边（≥2 个 model 节点）",
            "vlm_target_node_selected": "fct_customer_orders 或 customer_orders 节点被选中/高亮",
            "vlm_column_panel": "右侧或浮层中能看到 columns 列表，至少含 amount 字段",
            "vlm_lineage_arrows_distinct": "上游模型用不同颜色或不同标识的边连接（不是单一同色）",
        }
        try:
            vlm = vlm_score_rubric(vlm_imgs[:3], rubric, instruction="评估 dbt-docs 列血缘 GUI 取证截图。")
            for k in rubric: s[k] = float(vlm.get(k, 0.0))
            s["judge_method"] = vlm.get("judge_method","ok")
        except Exception:
            pass

    # ---- aggregate + hard gates ----
    nums = [v for v in s.values() if isinstance(v,(int,float))]
    base = sum(nums)/len(nums) if nums else 0.0
    has_cli_evidence = (s["run_before_log"] + s["test_before_log_fail"]
                        + s["build_after_log_pass"] + s["manifest_deps_jq_extract"]
                        + s["column_types_present"]) >= 4.0
    if not has_cli_evidence: base = min(base, 0.4)
    # NOTE: We deliberately do NOT cap on missing GUI screenshots.
    # Whether the agent invokes GUI tooling is not a scoring axis.
    # Missing PNGs already cost the corresponding sub_scores
    # (gui_screenshots_count / gui_screenshots_ocr / 4 vlm_*); no
    # additional hard cap is applied here.
    # VLM cap uses the rubric mean (consistent with batch3) and is only
    # enforced when the helper actually ran.
    if vlm_score_rubric is not None and vlm_imgs:
        vlm_avg = sum(s.get(k, 0.0) for k in
            ["vlm_dag_visible","vlm_target_node_selected",
             "vlm_column_panel","vlm_lineage_arrows_distinct"]) / 4.0
        if vlm_avg < 0.5:
            base = min(base, 0.6)
    if s["build_after_log_pass"] < 1.0 or s["duckdb_schema_fixed"] < 1.0:
        base = min(base, 0.45)
    if s["fixed_models_has_cast"] < 1.0:
        base = min(base, 0.45)
    s["overall_score"] = round(base, 3)
    return s
```

## Workspace Path

```
workspace/DAV/task_11_dbt_lineage_docs_cycle
```

## Skills

```
```

## Env

```
```

## Warmup

```bash
mkdir -p /tmp_workspace/results /tmp_workspace/state /opt/dav11_gt
which jq || (apt-get update -qq && DEBIAN_FRONTEND=noninteractive apt-get install -y -qq python3-pip git jq tesseract-ocr poppler-utils curl)
pip3 install --break-system-packages --quiet 'dbt-core>=1.7,<1.9' 'dbt-duckdb>=1.7' 'duckdb>=0.10' pytesseract pillow 2>/dev/null || pip3 install --quiet 'dbt-core>=1.7,<1.9' 'dbt-duckdb>=1.7' 'duckdb>=0.10' pytesseract pillow || true
test -d /tmp_workspace/jaffle_shop || (cp -r /tmp_workspace/exec/jaffle_shop /tmp_workspace/jaffle_shop 2>/dev/null || true)
if [ ! -d /tmp_workspace/jaffle_shop/models ]; then rm -f /tmp_workspace/state/jaffle.tgz; ok=0; for i in $(seq 1 3); do for url in "https://github.com/dbt-labs/jaffle_shop/archive/refs/tags/v0.3.0.tar.gz" "https://ghproxy.net/https://github.com/dbt-labs/jaffle_shop/archive/refs/tags/v0.3.0.tar.gz" "https://gh-proxy.com/https://github.com/dbt-labs/jaffle_shop/archive/refs/tags/v0.3.0.tar.gz" "https://mirror.ghproxy.com/https://github.com/dbt-labs/jaffle_shop/archive/refs/tags/v0.3.0.tar.gz"; do timeout 90 curl -fsSL --max-time 60 --connect-timeout 15 -o /tmp_workspace/state/jaffle.tgz "$url" 2>/dev/null && ok=1 && break; done; if [ "$ok" = "1" ]; then break; fi; sleep 3; done; if [ -s /tmp_workspace/state/jaffle.tgz ]; then tar -xzf /tmp_workspace/state/jaffle.tgz -C /tmp_workspace/state 2>/dev/null && rm -rf /tmp_workspace/jaffle_shop && mv /tmp_workspace/state/jaffle_shop-* /tmp_workspace/jaffle_shop 2>/dev/null || true; fi; fi || true
if [ -d /tmp_workspace/exec/overlay ]; then cp -r /tmp_workspace/exec/overlay/* /tmp_workspace/jaffle_shop/ 2>/dev/null || true; fi
if [ -d /tmp_workspace/overlay ]; then cp -r /tmp_workspace/overlay/* /tmp_workspace/jaffle_shop/ 2>/dev/null || true; fi
mkdir -p /root/.dbt
cp /tmp_workspace/jaffle_shop/profiles.yml /root/.dbt/profiles.yml 2>/dev/null || true
cd /tmp_workspace/jaffle_shop && dbt deps >/tmp/dbt_deps.log 2>&1 || true
cd /tmp_workspace/jaffle_shop && dbt seed --quiet >/tmp/dbt_seed.log 2>&1 || true
# Move the agent-visible gt/expected.json to a root-only path so it can
# only be read by the grader (which runs as root). The agent must
# discover failing_model / conflicting_column / upstream types by
# running dbt and inspecting duckdb, not by reading the answer file.
if [ -f /tmp_workspace/gt/expected.json ]; then
  mv /tmp_workspace/gt/expected.json /opt/dav11_gt/expected.json 2>/dev/null || true
  chown -R root:root /opt/dav11_gt 2>/dev/null || true
  chmod 700 /opt/dav11_gt 2>/dev/null || true
  chmod 600 /opt/dav11_gt/expected.json 2>/dev/null || true
  rmdir /tmp_workspace/gt 2>/dev/null || true
fi
echo "[warmup] jaffle_shop ready at /tmp_workspace/jaffle_shop"
```