import dlt

@dlt.source
def my_source():
    @dlt.resource
    def events():
        yield [
            {"id": 1, "updated_at": 1},
            {"id": 2, "updated_at": 2},
            {"id": 3, "updated_at": 3},
        ]
    return events

def count_rows(items, meta, metrics):
    if isinstance(items, list):
        metrics["row_count"] = metrics.get("row_count", 0) + len(items)

source = my_source()
r = source.events  # _args_bound is False

r.apply_hints(incremental=dlt.sources.incremental("updated_at", initial_value=2))
r.add_metrics(count_rows, insert_at=len(r._pipe))
# Steps: [gen, Incremental(affinity=1), MetricsItem(affinity=0)]

pipeline = dlt.pipeline(pipeline_name="test", destination="duckdb", dev_mode=True)
pipeline.run(r)

for step in pipeline.last_trace.steps:
    if step.step == "extract":
        for load_id, metrics_list in step.step_info.metrics.items():
            for m in metrics_list:
                for resource_name, resource_m in m["resource_metrics"].items():
                    if resource_m.custom_metrics:
                        print(f"{resource_name}: {resource_m.custom_metrics}")

# Expected: row_count == 2  (2 rows pass incremental: updated_at >= 2)
# Actual:   row_count == 3  (all rows — MetricsItem ran BEFORE incremental)