Waterfall Engine and Scaling Policy

[gxuu]

READY FOR REVIEW.

Waterfall_v1 engine and Waterfall Scaling Policy

Summary

Adds a new waterfall scaling policy that cascades worker scaling across prioritized worker managers. Priority-1 managers fill first; overflow goes to priority-2, then priority-3. Shutdown is reversed — least-preferred drains first. Offline managers are skipped automatically.

Configuration

policy_engine_type = "waterfall_v1"
policy_content = """
# priority, worker_type, max_task_concurrency
         1,      native,                   10
         1,     native1,                   20
         2,         ecs,                   64
"""

Each line is priority,worker_type,max_task_concurrency. Comments (#) and blank lines are supported. Worker types match worker manager IDs by prefix (e.g. worker_type native matches manager ID native|abc123). Effective capacity per manager is min(config max_task_concurrency, heartbeat max_worker_groups).

Changes

Protocol:

• Added workerManagerID (Data) to WorkerManagerHeartbeat in Cap'n Proto schema (required, no default)
• Each manager generates its own ID as <worker_type>|<uuid>: NAT|<pid>, ECS|<pid>, SYM|<pid>

Scaling interface:

• Extended ScalingPolicy.get_scaling_commands() with a worker_manager_snapshots parameter (Dict[bytes, WorkerManagerSnapshot]) providing cross-manager state to all policies
• WorkerManagerController builds these snapshots from existing heartbeat tracking state via _build_manager_snapshots()
• Added WorkerManagerSnapshot dataclass to simple_policy/scaling/types.py

Policy layer:

• WaterfallV1Policy implements ScalerPolicy (same as SimplePolicy), routed through VanillaPolicyController via create_policy()
• VanillaPolicyController remains the only PolicyController implementation
• Added WATERFALL_V1 case to create_policy() factory in library/utility.py

New files:

• waterfall_v1/waterfall_v1_policy.py — WaterfallV1Policy(ScalerPolicy) with even_load allocation
• waterfall_v1/scaling/waterfall.py — Stateless WaterfallScalingPolicy(ScalingPolicy) with worker_type-based manager ID matching
• waterfall_v1/scaling/types.py — WaterfallRule (fields: priority, worker_type, max_task_concurrency)
• waterfall_v1/scaling/utility.py — parse_waterfall_rules() config parser
• tests/scheduler/test_waterfall_scaling.py — 25 unit tests

[yzard]

@gxuu, the name should be WaterfallPolicyEngine, it should not be controller level

[magniloquency]

minor