---
id: DAV_task_10_polars_streamlit_xfilter
name: Polars 惰性计划 × Streamlit-Altair 刷选定位长尾分区
category: DAV
timeout_seconds: 1800
---

## Prompt
The data team maintains a Polars LazyFrame ETL that aggregates BTS flight on-time data by `origin × carrier` to produce p95 delay and cancellation rate. Most batches finish in 30s, but a few take 4–6 minutes. **The problem is not "find a single bottleneck"** — the same pipeline has multiple kinds of bottlenecks coexisting across different abstraction layers (logical plan / physical exec / parquet storage), and at least one anomalous-looking metric spike is actually a measurement artifact and must not be listed as a root cause.

Materials already in place:
- Input data: `/tmp_workspace/data/flights.parquet` (≈ 600k rows of synthetic data following the BTS schema; carries row-group metadata, so `pyarrow.parquet.ParquetFile` can extract `num_row_groups`, plus per-row-group `num_rows` / column chunk byte sizes).
- Baseline query script: `/tmp_workspace/etl/query.py` (a production-equivalent lazy query; `--explain` / `--profile` print the Polars optimized plan and a per-node timing table respectively; `--repeat N` runs the same query repeatedly to surface JIT/GC/cache-class measurement artifacts).
- Debug dashboard: `/tmp_workspace/etl/streamlit_app.py`, automatically started by warmup and listening on `http://localhost:8501`. It includes a top KPI strip, a Departures by Hour histogram, an Altair scatter plot (origin frequency × p95 delay), and a linked datatable. The scatter supports a rectangular brush selection; once a brush is active the linked table shows only the selected `(origin, carrier)` combinations; hovering on a row pops up a tooltip with the true row count (**this number only exists in the front-end runtime state** — it cannot be obtained from the static DOM or backend JSON, so a screenshot of the rendered state is required).

Task deliverables (drop artifacts into `/tmp_workspace/results/`):

1. **Execution-plan evidence (before / after)**
   - `plan_before.txt` / `plan_after.txt`: the optimized plan text of the original and rewritten queries; each must contain at least two keywords drawn from `JOIN` / `FILTER` / `SCAN` / `AGGREGATE`; in `plan_after.txt` the filter predicate must appear before the join (predicate pushdown).

2. **Per-node profile evidence (before / after)**
   - `profile_before.txt` / `profile_after.txt`: the `.profile()` output (header + body). `profile_before.txt` must retain ≥ 8 node-record lines (used for event-timing reasoning).

3. **Rewritten query source** `optimized_query.py`
   - Must be parseable by `ast.parse`; must `import polars`; in source-character order `.filter(...)` must appear before `.join(...)`; semantically equivalent to `query.py` (hard-coding to bypass the skewed partitions is forbidden). The top of the file must contain a `# COUNTER-EXAMPLES` comment block listing ≥ 2 rewrite attempts that were tried but did not work, with their failure reasons (e.g. "switching the join to a cross-join caused OOM").

4. **Long-tail partition list** `skew_partitions.json`
   - JSON array with **≥ 7 entries**; per-entry schema:
     ```json
     {"origin":"<3-letter IATA>","carrier":"<2-letter>","p95_delay_min":<float>,"n_rows":<int>,"layer_evidence":["logical_plan","physical_exec","storage"]}
     ```
   - `layer_evidence` must contain ≥ 2 items (i.e. each skew must be supported by evidence from at least two abstraction layers). All values must come from the real ETL run and the rendered dashboard — fabrication is forbidden.

5. **Multi-bottleneck causal chain** `bottlenecks.json`
   - JSON array with **≥ 5 entries**; each entry represents a bottleneck of a different `type` (allowed types: `join_explosion` / `projection_bloat` / `cardinality_blowup` / `late_filter` / `sort_skew` / `groupby_spill`; **the 5 entries must all use different types**). Per-entry schema:
     ```json
     {
       "id":"B1",
       "type":"join_explosion",
       "abstraction_layers":["logical_plan","physical_exec","storage"],
       "causal_chain":["plan_node X selectivity 1.0","intermediate result inflates to 4.2M rows after the join","physical operator spills to disk","wall-clock +210s"],
       "fix":"...",
       "counter_example":"...",
       "benchmark_before_ms":<float>,
       "benchmark_after_ms":<float>
     }
     ```
   - `abstraction_layers` ≥ 2 items; `causal_chain` ≥ 4 hops; `benchmark_before_ms > benchmark_after_ms`.

6. **Cross-source correlation** `cross_source_evidence.json`
   - Tie each bottleneck to ≥ 3 independent evidence sources. Schema:
     ```json
     {
       "links":[
         {"bottleneck_id":"B1","plan_node_id":"<from plan_after.txt>","profile_node_time_ms":<float>,"dashboard_n_rows":<int>,"parquet_row_group_count":<int>},
         ...
       ]
     }
     ```
   - The `links` array must have ≥ 5 entries; `plan_node_id` must actually appear in `plan_after.txt` (substring match).

7. **Multi-layer abstraction evidence** `multilayer_evidence.json`
   - For each bottleneck id, provide concrete evidence (snippet + numeric value) from at least 3 abstraction layers:
     ```json
     {
       "B1":{
         "logical_plan":{"snippet":"FILTER ... ON ...","node_id":"..."},
         "physical_exec":{"snippet":"profile row","time_ms":<float>},
         "storage":{"row_group_id":<int>,"row_group_rows":<int>,"column_chunk_bytes":<int>}
       },
       ...
     }
     ```
   - Must cover at least the first 3 ids in `bottlenecks.json`.

8. **Event-timing reasoning** `event_timeline.json`
   - A JSON array of ≥ 8 events sorted by `t_start_ms`; each entry:
     ```json
     {"step":<int>,"node_id":"<plan node>","t_start_ms":<float>,"t_end_ms":<float>,"parent":"<parent node_id or null>"}
     ```
   - Must allow a root → leaf causal chain to be reconstructed (the root's parent is null; every other parent must reference a node_id that appears earlier in the array). The timeline must also show that "filter pushed before join" is visible after the rewrite (filter step.t_start_ms < join step.t_start_ms).

9. **False-signal identification** `false_positives.json`
   - A JSON array with ≥ 1 entry; each entry:
     ```json
     {"metric":"<name>","observed_spike":"<description>","claimed_root_cause":"<misleading conclusion>","actual_explanation":"<true cause, e.g. JIT warmup / GC / page cache>","is_artifact":true}
     ```
   - Must be derived from multiple `--repeat` runs; at least one entry with `is_artifact:true`. The same metric **must not** also appear in `bottlenecks.json`.

10. **Formal equivalence verification** `equivalence.json`
    - Run `query.py` and `optimized_query.py` on the same input, sort the final aggregated result by `(origin, carrier)`, and compute the sha256:
      ```json
      {"hash_before":"<sha256>","hash_after":"<sha256>","equal":true,"row_count_before":<int>,"row_count_after":<int>}
      ```
    - Requires `equal=true` and `hash_before == hash_after` and `row_count_before == row_count_after`.

11. **Dashboard visual evidence** (each ≥ 4 KB, captured with pyautogui or gnome-screenshot)
    - `view_streamlit_overview.png`: full first screen of the dashboard (KPIs / histogram / scatter all in frame).
    - `view_brush_tooltip.png`: rectangular brush selection over the long-tail cluster on the scatter plot, with the hover tooltip overlay (including the `n=` number).
    - `view_brush_table.png`: the datatable linked to the brush selection, with ≥ 7 rows including the columns `origin` / `carrier` / `p95_delay_min` / `n_rows`.
    - `view_brush_after.png`: after rerunning the rewritten query, the same brushed region reproduced on the dashboard.

12. **Final report** `findings.json`
    ```json
    {
      "total_runtime_before_s":<float>,
      "total_runtime_after_s":<float>,
      "speedup":<float>,
      "bottleneck_count":<int>,
      "false_positive_id":"<name from false_positives.json>",
      "cross_source_chain_ids":["B1","B2","..."],
      "equivalence_hash":"<sha256>",
      "skew_dimension":"<column>",
      "skew_values":["<v1>","..."],
      "tail_share_pct":<float>,
      "predicate_pushed_down":"<verbatim snippet from optimized_query.py>",
      "evidence_screenshots":["view_streamlit_overview.png","view_brush_tooltip.png","view_brush_table.png","view_brush_after.png"]
    }
    ```

Hard constraints (cross-dependent — failing any one of them auto-zeroes downstream sub-scores):
- `total_runtime_before_s / total_runtime_after_s ≥ 5.0` and `speedup ≥ 4.5`.
- `bottlenecks.json` must contain ≥ 5 distinct `type` values; count < 5 → multi-layer-evidence / cross-source-evidence sub-scores auto-zero.
- `equivalence.json.equal` must be `true`; mismatch → speedup and predicate sub-scores auto-zero.
- A metric marked `is_artifact:true` in `false_positives.json` **must not** appear in any `causal_chain` string in `bottlenecks.json`; if it does, trap_score=0 and the overall score is capped at 0.4.
- In `event_timeline.json` the filter node's `t_start_ms` must be < the join node's `t_start_ms`, otherwise predicate_pushdown_ordering auto-zeroes.
- All numeric values must come from real runs / real parquet metadata / the real rendered dashboard state; hard-coding or bypassing the skew is forbidden.

## Expected Behavior

合格交付应在 `/tmp_workspace/results/` 留下一条**多层 × 多源 × 可时序还原**的取证链，使第三方仅凭文件即可独立复核：

- 一对计划文件 (`plan_before.txt` / `plan_after.txt`) 让审阅者一眼看出 filter 谓词被前移到 join 之前；plan_after 中的 plan_node_id 必须是 `cross_source_evidence.json` 中 `links[*].plan_node_id` 的真实子串来源。
- 一对 profile 文件 (`profile_before.txt` / `profile_after.txt`)，前者保留 ≥ 8 条节点记录用于事件时序推理；累加耗时与 `findings.json.total_runtime_*` 自洽，速比 ≥ 5×。
- `optimized_query.py` 可被 `ast.parse` 解析；源码字符序 `.filter(...)` 早于 `.join(...)`；顶部 `# COUNTER-EXAMPLES` 注释列 ≥ 2 个被否定的尝试方案及失败原因。
- `skew_partitions.json` ≥ 7 条；每条 `layer_evidence` 列至少 2 个抽象层。
- `bottlenecks.json` ≥ 5 条且 type 互不相同；每条含 ≥ 4 跳 `causal_chain` 与 ≥ 2 个 `abstraction_layers`，且 `benchmark_before_ms > benchmark_after_ms`。
- `cross_source_evidence.json.links` ≥ 5 条，把 plan_node / profile 时间 / 看板行数 / parquet row group 四源关联；`plan_node_id` 必须是 `plan_after.txt` 中真实子串。
- `multilayer_evidence.json` 覆盖前 3 个 bottleneck id，每个 id 给出 logical_plan / physical_exec / storage 三层的具体 snippet 与数值。
- `event_timeline.json` ≥ 8 条事件按时间排序，且任一非 root 事件的 `parent` 必须指向数组中较早出现的 `node_id`；filter 节点 `t_start_ms` < join 节点 `t_start_ms`。
- `false_positives.json` 至少识别 1 个 measurement artifact (`is_artifact:true`)，且该 metric 不出现在 `bottlenecks.json` 任何 `causal_chain` 字符串中——即 agent 必须**主动否定**这个伪信号而非把它当成根因。
- `equivalence.json` 给出 before/after 聚合结果 sha256，`equal:true` 且 row_count 一致，证明改写未破坏语义。
- 四张看板截图 (overview / brush+tooltip / brush+table≥7 行 / brush_after) 由 `pyautogui.screenshot()` 或 `gnome-screenshot` 抓取；VLM 可识别 brush 矩形 + tooltip 浮层 (含 `n=`)。
- `findings.json` 字段齐全 (含 `bottleneck_count` / `false_positive_id` / `cross_source_chain_ids` / `equivalence_hash`)，作为整套取证的索引。

任何一条主链断裂 (例如 false_positive 写成根因 / equivalence 哈希不一致 / bottleneck < 5) 都会触发 grader 的 cascading penalty，使总分被 cap 在 0.4 附近。

## Source

- BTS On-Time Reporting Departures schema (https://www.transtats.bts.gov/ONTIME/Departures.aspx)
- Polars docs: https://docs.pola.rs/user-guide/lazy/optimizations/
- Streamlit + Altair brush selection demo: https://altair-viz.github.io/gallery/scatter_linked_table.html
- 真实 reddit 讨论（Polars lazy plan 看着对但跑得慢）：https://www.reddit.com/r/Python/comments/1ax... (Polars perf threads)
- 对应 benchmark: D10

## Grading Criteria

- [ ] 1. plan_before.txt / plan_after.txt 存在且各含 ≥ 2 类 plan 关键字 (JOIN/FILTER/SCAN/AGGREGATE)
- [ ] 2. profile_before.txt 含 ≥ 8 行节点记录；profile_after.txt 含表头与表体
- [ ] 3. optimized_query.py 能 ast.parse；含 `import polars`；源码字符序 .filter < .join；顶部含 `# COUNTER-EXAMPLES` 注释块
- [ ] 4. 加速比 (before/after) ≥ 5.0 且 findings.speedup ≥ 4.5
- [ ] 5. skew_partitions.json ≥ 7 行且 schema 含 layer_evidence (≥ 2 项)
- [ ] 6. bottlenecks.json ≥ 5 条且 type 互不相同；每条 causal_chain ≥ 4 跳且 abstraction_layers ≥ 2
- [ ] 7. cross_source_evidence.json links ≥ 5 条；每条含 plan_node_id / profile_node_time_ms / dashboard_n_rows / parquet_row_group_count；plan_node_id 子串真实出现于 plan_after.txt
- [ ] 8. multilayer_evidence.json 覆盖 ≥ 3 个 bottleneck id，每 id 三层 (logical_plan/physical_exec/storage) 都给出 snippet+数值
- [ ] 9. event_timeline.json ≥ 8 条按 t_start_ms 排序；非 root 的 parent 指向更早 node_id；filter 节点 t_start_ms < join 节点 t_start_ms
- [ ] 10. false_positives.json ≥ 1 条 is_artifact:true，且其 metric 名不出现于 bottlenecks.json 任一 causal_chain
- [ ] 11. equivalence.json 含 hash_before/hash_after/equal/row_count_before/row_count_after，equal=true 且哈希一致
- [ ] 12. findings.json 字段齐全 (含 bottleneck_count / false_positive_id / cross_source_chain_ids / equivalence_hash / predicate_pushed_down)，predicate_pushed_down 子串真实出现于 optimized_query.py
- [ ] 13. 4 张 GUI 截图齐全，OCR 命中关键 UI 字 (`Departures`/`p95`/`carrier`/`n=`)
- [ ] 14. VLM 判定 brush 截图含 brush 矩形 + tooltip 浮层 + 联动表格 ≥ 7 行

## Automated Checks

```python
def grade(workspace_path=None, **kwargs) -> dict:
    """Multi-dim grader: plans + profile + speedup + skew JSON + GUI shots + VLM rubric.
    Hard gates: missing CLI evidence -> 0.4, missing GUI shots -> 0.4, VLM<0.6 -> 0.6."""
    import json, re, ast
    from pathlib import Path
    rd = Path("/tmp_workspace/results")
    # Read GT from root-only path (warmup moves it off /tmp_workspace/gt
    # to prevent agent from grepping skew_carriers / notes solution map).
    gt_dir = Path("/opt/dav10_gt") if Path("/opt/dav10_gt/expected.json").exists() else Path("/tmp_workspace/gt")
    s = {}

    # ---- 1. plan / profile artefacts ----
    plan_b = rd / "plan_before.txt"
    plan_a = rd / "plan_after.txt"
    prof_b = rd / "profile_before.txt"
    prof_a = rd / "profile_after.txt"
    s["plan_before_exists"]    = 1.0 if plan_b.exists() and plan_b.stat().st_size > 30 else 0.0
    s["plan_after_exists"]     = 1.0 if plan_a.exists() and plan_a.stat().st_size > 30 else 0.0
    # Prompt 第 1 项 + bullet 2: profile_before.txt 必须 ≥ 8 行节点记录
    # (用于事件时序推理). 计算非空行数；少于 8 行视为占位文件，不给分.
    def _profile_lines(p):
        if not p.exists(): return 0
        try:
            return sum(1 for ln in p.read_text(errors="ignore").splitlines() if ln.strip())
        except Exception:
            return 0
    pb_lines = _profile_lines(prof_b)
    s["profile_before_exists"] = 1.0 if (prof_b.exists() and prof_b.stat().st_size > 30 and pb_lines >= 8) else \
                                 (0.5 if (prof_b.exists() and prof_b.stat().st_size > 30) else 0.0)
    s["profile_after_exists"]  = 1.0 if prof_a.exists() and prof_a.stat().st_size > 30 else 0.0

    plan_keywords = ["JOIN", "FILTER", "AGGREGATE", "SCAN", "SELECTION"]
    def has_plan_kw(p):
        if not p.exists(): return 0.0
        try:
            t = p.read_text(errors="ignore").upper()
            return 1.0 if sum(1 for k in plan_keywords if k in t) >= 2 else 0.0
        except Exception:
            return 0.0
    s["plan_before_has_keywords"] = has_plan_kw(plan_b)
    s["plan_after_has_keywords"]  = has_plan_kw(plan_a)

    # ---- 2. predicate pushed down before join in optimized_query.py ----
    opt = rd / "optimized_query.py"
    s["optimized_exists"] = 1.0 if opt.exists() else 0.0
    pushdown_ok = 0.0
    src_text = ""
    if opt.exists():
        try:
            src_text = opt.read_text(errors="ignore")
            ast.parse(src_text)
            s["optimized_parses"] = 1.0
            # bullet #3: file head must contain a `# COUNTER-EXAMPLES`
            # comment block listing ≥ 2 rejected attempts. Look in the
            # first ~80 lines for the marker plus at least 2 bullet items.
            head = "\n".join(src_text.splitlines()[:80])
            if "# COUNTER-EXAMPLES" in head:
                # count bullet-style `# - …` or `# * …` lines after marker
                tail = head.split("# COUNTER-EXAMPLES", 1)[1]
                bullets = sum(
                    1 for ln in tail.splitlines()
                    if re.match(r"\s*#\s*[-*0-9]", ln)
                )
                s["optimized_counter_examples"] = 1.0 if bullets >= 2 else (0.5 if bullets >= 1 else 0.0)
            else:
                s["optimized_counter_examples"] = 0.0
            # Heuristic: filter() call appearing before join() call in source order
            tree = ast.parse(src_text)
            chains = []
            for node in ast.walk(tree):
                if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute):
                    chains.append((node.func.attr, node.lineno, node.col_offset))
            f_pos = [(l,c) for (a,l,c) in chains if a == "filter"]
            j_pos = [(l,c) for (a,l,c) in chains if a == "join"]
            joined_before_filter = any((jl,jc) < (fl,fc) for (fl,fc) in f_pos for (jl,jc) in j_pos)
            if f_pos and j_pos and not joined_before_filter and "scan_parquet" in src_text:
                pushdown_ok = 1.0
        except Exception:
            s["optimized_parses"] = 0.0
    else:
        s["optimized_parses"] = 0.0
    s["predicate_pushdown_ordering"] = pushdown_ok

    # ---- 3. skew_partitions.json ----
    sp = rd / "skew_partitions.json"
    sp_rows = []
    if sp.exists():
        try:
            sp_rows = json.loads(sp.read_text())
        except Exception:
            sp_rows = []
    sp_schema_ok = (
        isinstance(sp_rows, list) and len(sp_rows) >= 7
        and all(isinstance(r, dict)
                and {"origin","carrier","p95_delay_min","n_rows","layer_evidence"} <= set(r.keys())
                and isinstance(r.get("n_rows"), int)
                and isinstance(r.get("layer_evidence"), list) and len(r["layer_evidence"]) >= 2
                for r in sp_rows)
    )
    s["skew_partitions_schema"] = 1.0 if sp_schema_ok else 0.0

    expected = {}
    if (gt_dir/"expected.json").exists():
        try: expected = json.loads((gt_dir/"expected.json").read_text())
        except Exception: pass

    skew_carrier_hit = 0.0
    if sp_schema_ok and "skew_carriers" in expected:
        cs = {r.get("carrier") for r in sp_rows}
        gold = set(expected["skew_carriers"])
        if cs & gold:
            skew_carrier_hit = 1.0
    s["skew_partition_value_match"] = skew_carrier_hit

    # ---- 4. findings.json ----
    fnd = rd / "findings.json"
    fnd_data = {}
    if fnd.exists():
        try: fnd_data = json.loads(fnd.read_text())
        except Exception: pass
    needed = ["total_runtime_before_s","total_runtime_after_s","speedup",
              "skew_dimension","skew_values","tail_share_pct",
              "predicate_pushed_down","evidence_screenshots",
              # bullet #12 — surface the four schema fields that were
              # silently unenforced in earlier batch3 builds.
              "bottleneck_count","false_positive_id",
              "cross_source_chain_ids","equivalence_hash"]
    s["findings_schema"] = 1.0 if all(k in fnd_data for k in needed) else \
                           (sum(1 for k in needed if k in fnd_data) / len(needed))

    speedup_ok = 0.0
    try:
        sp_val = float(fnd_data.get("speedup", 0))
        rb = float(fnd_data.get("total_runtime_before_s", 0))
        ra = float(fnd_data.get("total_runtime_after_s", 0))
        if rb >= 5.0 and ra > 0 and (rb/ra) >= 4.5 and sp_val >= 4.5:
            speedup_ok = 1.0
        elif rb >= 3.0 and ra > 0 and (rb/ra) >= 3.0:
            speedup_ok = 0.5
        elif rb > 0 and ra > 0 and (rb/ra) >= 2.0:
            speedup_ok = 0.25
    except Exception:
        pass
    s["speedup_ge_2x"] = speedup_ok

    # predicate text actually appears in optimized_query.py
    pred_text = str(fnd_data.get("predicate_pushed_down","")).strip()
    s["predicate_text_in_source"] = 1.0 if (
        pred_text and len(pred_text) >= 20 and src_text and pred_text in src_text
        and "carrier" in pred_text.lower()
    ) else 0.0

    def _jload(name):
        p = rd/name
        try: return json.loads(p.read_text()) if p.exists() else None
        except Exception: return None
    bn = _jload("bottlenecks.json") or []
    fp = _jload("false_positives.json") or []
    eq = _jload("equivalence.json") or {}
    cs = _jload("cross_source_evidence.json") or {}
    ml = _jload("multilayer_evidence.json") or {}
    et = _jload("event_timeline.json") or []
    types = {b.get("type") for b in bn if isinstance(b, dict)}
    s["bottlenecks_5_distinct_types"] = 1.0 if len(types) >= 5 else 0.0
    s["bottlenecks_causal_chain_ok"]  = 1.0 if bn and all(
        isinstance(b,dict) and len(b.get("causal_chain",[]))>=4
        and len(b.get("abstraction_layers",[]))>=2 for b in bn) else 0.0
    s["equivalence_hash_match"] = 1.0 if eq.get("equal") is True and eq.get("hash_before") and eq.get("hash_before")==eq.get("hash_after") else 0.0
    plan_after_txt = plan_a.read_text(errors="ignore") if plan_a.exists() else ""
    links = cs.get("links",[]) if isinstance(cs,dict) else []
    s["cross_source_links_ok"] = 1.0 if len(links)>=5 and all(
        isinstance(l,dict) and l.get("plan_node_id") and str(l["plan_node_id"]) in plan_after_txt
        for l in links) else 0.0
    s["multilayer_three_layers"] = 1.0 if isinstance(ml,dict) and len(ml)>=3 and all(
        {"logical_plan","physical_exec","storage"} <= set(v.keys()) for v in list(ml.values())[:3]) else 0.0
    s["event_timeline_ordered"] = 0.0
    if isinstance(et,list) and len(et)>=8:
        ts = [e.get("t_start_ms",0) for e in et]
        ids = {e.get("node_id") for e in et}
        order_ok = ts == sorted(ts) and all((e.get("parent") in ids) or e.get("parent") in (None,"null") for e in et)
        f_t = min((e["t_start_ms"] for e in et if "filter" in str(e.get("node_id","")).lower()), default=None)
        j_t = min((e["t_start_ms"] for e in et if "join"   in str(e.get("node_id","")).lower()), default=None)
        if order_ok and f_t is not None and j_t is not None and f_t < j_t:
            s["event_timeline_ordered"] = 1.0
    fp_artifact_metrics = {str(x.get("metric","")).lower() for x in fp if isinstance(x,dict) and x.get("is_artifact")}
    leak = any(m and any(m in " ".join(b.get("causal_chain",[])).lower() for b in bn) for m in fp_artifact_metrics)
    s["false_positive_isolated"] = 1.0 if fp_artifact_metrics and not leak else 0.0

    # ---- 5. GUI screenshots ----
    gui_shots = ["view_streamlit_overview.png","view_brush_tooltip.png",
                 "view_brush_table.png","view_brush_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.5
    try:
        import pytesseract
        from PIL import Image
        kws = {
            "view_streamlit_overview.png": ["Departures","Hour","p95","delay","origin"],
            "view_brush_tooltip.png":      ["n=","carrier","p95","delay"],
            "view_brush_table.png":        ["origin","carrier","p95","n_rows"],
            "view_brush_after.png":        ["carrier","p95","origin"],
        }
        hits = 0
        for n, ks in kws.items():
            p = rd/n
            if p.exists():
                try:
                    tx = pytesseract.image_to_string(Image.open(p))
                    if any(k.lower() in tx.lower() for k in ks):
                        hits += 1
                except Exception:
                    pass
        gui_ocr = hits / len(kws)
    except ImportError:
        pass
    s["gui_screenshots_ocr"] = gui_ocr

    # ---- 6. VLM rubric (optional) ----
    try:
        from _judge_helper import vlm_score_rubric
    except Exception:
        vlm_score_rubric = None
    vlm_imgs = [str(rd/n) for n in
                ["view_brush_tooltip.png","view_brush_table.png","view_streamlit_overview.png"]
                if (rd/n).exists()]
    if vlm_score_rubric and vlm_imgs:
        rubric = {
          "vlm_brush_rect_visible": "散点图上必须同时看到一个半透明矩形 brush 选区，且选区内圆点保留原色、选区外的点变为浅灰",
          "vlm_tooltip_with_n":     "矩形 brush 之上叠加一个 hover tooltip 浮层，浮层文字明确包含 'n=' 后跟整数 (≥3 位数)",
          "vlm_filtered_table":     "页面下方 datatable 行数明显少于默认 20，且首列严格按 p95_delay_min 降序排列，包含数据中实际出现的 carrier 代码（agent 通过运行 etl 推断）",
          "vlm_layout_clean":       "整张截图同框完整呈现 4 个 KPI 卡 + Departures by Hour 直方图 + 散点 + datatable，无截断、无滚动条",
        }
        try:
            vlm = vlm_score_rubric(vlm_imgs[:3], rubric, instruction="评估 Polars 长尾分区调试 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["plan_before_exists"] + s["profile_before_exists"]
                        + s["plan_after_exists"] + s["profile_after_exists"]) >= 3.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 / vlm_*); no
    # additional hard cap is applied here.
    # VLM cap uses the rubric mean (consistent with other batch3 tasks)
    # 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_brush_rect_visible","vlm_tooltip_with_n",
             "vlm_filtered_table","vlm_layout_clean"]) / 4.0
        if vlm_avg < 0.5:
            base = min(base, 0.6)
    if s["speedup_ge_2x"] < 1.0 and s["predicate_pushdown_ordering"] < 1.0:
        base = min(base, 0.55)

    s["overall_score"] = round(base, 3)
    return s
```

## Workspace Path

```
workspace/DAV/task_10_polars_streamlit_xfilter
```

## Skills

```
```

## Env

```
```

## Warmup

```bash
mkdir -p /tmp_workspace/results /tmp_workspace/state /tmp_workspace/data /tmp_workspace/etl /tmp_workspace/gt /opt/dav10_priv /opt/dav10_gt
# Ensure the agent (non-root user) can write the rewrite (`optimized_query.py`)
# and supporting evidence into etl/results/state. Without this the warmup
# leaves these dirs root-owned and the agent cascades to 0 on the
# `optimized_*` and `predicate_*` sub-scores even when its analysis is
# correct.
chmod 777 /tmp_workspace/results /tmp_workspace/state /tmp_workspace/data /tmp_workspace/etl /tmp_workspace/gt 2>/dev/null || true
chown -R user:user /tmp_workspace/results /tmp_workspace/state /tmp_workspace/data /tmp_workspace/etl /tmp_workspace/gt 2>/dev/null || true
which tesseract || (apt-get update -qq && DEBIAN_FRONTEND=noninteractive apt-get install -y -qq python3-pip tesseract-ocr poppler-utils curl)
pip3 install --break-system-packages --ignore-installed --quiet blinker 2>/dev/null || pip3 install --ignore-installed --quiet blinker || true
pip3 install --break-system-packages --quiet 'polars>=0.20,<1.5' 'streamlit>=1.30' 'altair>=5' pandas numpy faker pyarrow pytesseract pillow 2>/dev/null || pip3 install --quiet 'polars>=0.20,<1.5' 'streamlit>=1.30' 'altair>=5' pandas numpy faker pyarrow pytesseract pillow
cp -r /tmp_workspace/exec/* /tmp_workspace/etl/ 2>/dev/null || true
for f in gen_data.py query.py streamlit_app.py requirements.txt README.md; do [ -f /tmp_workspace/$f ] && cp /tmp_workspace/$f /tmp_workspace/etl/ 2>/dev/null; done
mv /tmp_workspace/etl/gen_data.py /opt/dav10_priv/gen_data.py 2>/dev/null || true
chmod 700 /opt/dav10_priv && chown -R root:root /opt/dav10_priv
test -f /tmp_workspace/data/flights.parquet || python3 /opt/dav10_priv/gen_data.py --out /tmp_workspace/data/flights.parquet --rows 600000 --seed 17 >/tmp/gen_data.log 2>&1 || true
rm -f /opt/dav10_priv/gen_data.py
nohup streamlit run /tmp_workspace/etl/streamlit_app.py --server.port=8501 --server.headless=true --browser.gatherUsageStats=false >/tmp/streamlit.log 2>&1 &
sleep 3
for i in $(seq 1 30); do curl -sf http://localhost:8501/_stcore/health >/dev/null && break; sleep 1; done
# Move expected.json (contains skew_carriers + notes solution sketch) to
# root-only /opt/dav10_gt so the agent cannot grep the answer.
if [ -f /tmp_workspace/gt/expected.json ]; then
  mv /tmp_workspace/gt/expected.json /opt/dav10_gt/expected.json 2>/dev/null || true
  chown -R root:root /opt/dav10_gt 2>/dev/null || true
  chmod 700 /opt/dav10_gt 2>/dev/null || true
  chmod 600 /opt/dav10_gt/expected.json 2>/dev/null || true
  rmdir /tmp_workspace/gt 2>/dev/null || true
fi
```