MkDocs preview: https://cbed668e.dd-docs-preview.pages.dev

Fern preview: https://nvidia-preview-pr-685.docs.buildwithfern.com/nemo/datadesigner​

Notebook tutorials are rendered without execution outputs in previews.

Greptile Summary

This PR tightens two AIMD edge cases in AdaptiveRequestAdmissionController and wires up request-pressure advisory in the async scheduler. The core logic changes are in controller.py and are well-covered by three new focused tests.

• Stale-lease guard (controller.py line 365): the guard now compares the full frozen-dataclass lease for equality (active != lease) instead of only the item.resource field, so any caller that passes back a modified copy of the lease (e.g., via dataclasses.replace) is correctly rejected without corrupting in-flight counters.
• AIMD decrease gate (_apply_outcome): replaces first_in_cascade = (consecutive_rate_limits == 0) with should_decrease = (admitted_adaptive_limit <= prev_limit), using the limit snapshotted at acquisition time. This fixes a latent bug where, after the first burst cascade, a fresh rate-limit on a newly-admitted lease would never re-trigger a multiplicative decrease because consecutive_rate_limits was never 0 again. The rate_limit_ceiling is likewise anchored to the admitted limit rather than prev_limit, making the ceiling reflect the concurrency level the provider actually saw when it rate-limited.
• dataset_builder.py: adds request_pressure_advisory=True to the AsyncTaskScheduler constructor; a new monkeypatch integration test confirms the kwarg is forwarded.

Important Files Changed

sequenceDiagram
    participant Caller
    participant RAC as AdaptiveRequestAdmissionController
    participant State as AdaptiveRequestLimitState

    Caller->>RAC: try_acquire(item)
    RAC->>RAC: _acquire_locked(item, now)
    Note over RAC: snapshot current_limit → lease.current_adaptive_limit
    RAC-->>Caller: "RequestAdmissionLease (lease_id, current_adaptive_limit=N)"

    Caller->>RAC: release(lease, outcome)
    RAC->>RAC: "active = _active_leases.pop(lease_id)"
    alt "active != lease (stale: any field mismatch)"
        RAC->>RAC: "_active_leases[lease_id] = active  (restore)"
        RAC-->>Caller: "ReleaseResult(released=False, reason=stale_lease)"
    else exact match
        RAC->>RAC: _apply_outcome(state, resource, active.current_adaptive_limit, outcome, now, events)
        alt "outcome == rate_limited"
            RAC->>State: "consecutive_rate_limits += 1"
            RAC->>RAC: "should_decrease = (admitted_adaptive_limit <= state.current_limit)"
            alt should_decrease
                RAC->>State: "current_limit = floor(current_limit * MD_factor)"
                RAC->>State: "rate_limit_ceiling = min(ceiling, admitted_adaptive_limit)"
            end
        else "outcome == success (cooldown cleared)"
            RAC->>State: "consecutive_rate_limits = 0"
            RAC->>State: "success_streak += 1 → maybe increase current_limit"
        end
        RAC-->>Caller: "ReleaseResult(released=True, reason=released)"
    end

_{Reviews (2): Last reviewed commit: "revert request admission docs changes" | Re-trigger Greptile}