feat(planning): granular step-level execution planning for Stage.META

[shobhitagnihotri69]

• Route outstanding checks per episode directly to SelectedRegisteredSteps
• Treat episodes quarantined by settled critical checks as owing nothing in Stage.META
• Group episodes by owed steps in batch and direct stage execution
• Add tests for check addition, critical retuning, and batch grouping

Resolves #690

Summary

When new checks are registered or check versions/parameters are retuned, planning for Stage.META previously operated at a binary stage level: any unsettled check caused Stage.META to execute every check on every episode. Additionally, checks skipped downstream of a critical check quarantine never become settled in catalog history, which caused quarantined episodes to be continuously re-planned for Stage.META.

Per maintainer guidance in #690:

• Route each episode's OutstandingStages.outstanding_steps into the existing step-selection machinery (SelectedRegisteredSteps), with no second filter inside _process_steps.
• Episodes whose quarantine was produced by a settled critical check owe nothing more in Stage.META unless that critical check itself changes identity (retuning).
• Airflow sub-DAGs and direct execution keep the same granularity: group episodes by the step subset they owe into batches (step_names_by_uri), carrying step_names into process_stage_batch.

Changes

• src/hflow/stage_planning.py:
  • In plan_outstanding_stages, inspect episodes_latest for quarantine status (quarantined, quarantine_tags_json) and critical check identities.
  • If quarantined by a settled critical gate, mark the episode as owing nothing in Stage.META.
  • Record outstanding checks per episode in outstanding_steps.
  • In outstanding_stage_uris, support _stage_steps_by_uri to track steps owed per URI for the requested stage.
• src/hflow/stage_execution.py:
  • In PlannedStageBatch, add step_names: NotRequired[list[str] | None].
  • In plan_stage_batches, accept step_names_by_uri to partition URIs by identical required step subsets before batching.
  • In run_stages_directly, partition URIs by owed step names and pass _registered_step_selection=SelectedRegisteredSteps(frozenset(steps)).
• src/hflow/runtime/_templates.py:
  • In OUTSTANDING_PLAN_FILTER_TEMPLATE, collect stage_steps_by_uri and supply it to plan_stage_batches.
  • In process_batch, pass batch.get("step_names") to process_stage_batch(..., step_names=stage_step_names).
• Tests:
  • tests/test_stage_planning.py: Added TestGranularStepPlanningForMeta testing that adding a check only reruns that check on already-settled episodes and skips quarantined episodes, and that retuning a critical check re-evaluates the quarantined episode.
  • tests/test_stage_execution.py: Added test_batch_lane_groups_by_step_names to verify batch planning partitions items by owed step subsets.

Verification

• uv run ruff check & uv run ruff format --check passed cleanly (323 files formatted).
• Targeted test suites (test_runtime_deploy.py, test_stage_execution.py, test_stage_planning.py) passed.
• Full test suite passed: 2,477 passed, 49 skipped.

[greptile-apps]

Confidence Score: 4/5

[Medium risk] Adds step-level batch grouping to stage execution planning.

The PR is not ready to merge: quarantined episodes can still miss diagnostic checks.

Findings

1.  Quarantine blocks diagnostic checks ▶
2.  Step groups multiply batches ▶

Summary

Stage.META now plans work per check, so adding or retuning a check does not rerun checks already settled. Direct runs and Airflow batches carry each episode’s owed checks into execution.

• Stage.META runs only checks each episode still owes.
• Stage runners group episodes by the checks each one owes.

Diagram

%%{init: {'theme': 'neutral'}}%%
flowchart LR
  A[Catalog outcomes] --> B[Outstanding steps per episode]
  B --> C[Group episodes by owed steps]
  C --> D[Direct stage run]
  C --> E[Scheduled batches]

Loading

_{Reviews (3) · Last reviewed commit: "refactor(planning): address review feedb..."}

[greptile-apps]

Quarantine blocks diagnostic checks

When a settled critical check has quarantined an episode, this gate removes every check that still owes work. That includes default diagnostic checks, which the runner allows to run after quarantine. A default check that failed or was added later never records its evidence. Keep those checks in the plan.

Knowledge Base Used: Stage planning and execution

[greptile-apps]

Step groups multiply batches

batch_count now applies to each owed-step group rather than the whole stage. With batch_count=1, three step groups create three worker tasks; the default limit of four can also be exceeded. This adds workers and stagger delay beyond what the caller requested. Divide the stage-wide batch allowance among the groups.

Knowledge Base Used: Stage planning and execution

[kstonekuan]

Thanks, the approach matches the spec and the quarantine rule is right. Three structural changes before I merge:

• Template: step_names_by_uri=locals().get("stage_steps_by_uri") depends on whether the filter snippet was spliced in. Set stage_steps_by_uri = None at the top of plan() and pass it directly.
• plan_stage_batches: the new branch copies the file-size lookup and bin-packing block. Size the URIs once in the existing path, then bin-pack each step group with the same code, so there is one sizing path and one stagger rule.
• One owner for the per-stage step filter: run_stages_directly and outstanding_stage_uris both derive "outstanding steps for this stage" from plan.outstanding_steps and _registered_step_stages(). Put that in one helper in stage_planning.py and call it from both.

[shobhitagnihotri69]

Thanks for the review @kstonekuan,Addressed all three structural points:

1. Template: Set stage_steps_by_uri = None at the top of plan() and passed it directly to plan_stage_batches.
2. plan_stage_batches: Unified the URI sizing and bin-packing path so all URIs are sized once up front, and both single and multi-group cases share the exact same bin-packing logic and stagger rules.
3. Single owner for per-stage step filtering: Extracted outstanding_steps_for_stage(plan, stage, stage_by_step_name) in src/hflow/stage_planning.py and called it from both outstanding_stage_uris and run_stages_directly.

[kstonekuan]

LGTM, thanks for the quick rework. Full suite passes merged onto main. Merging.

[shobhitagnihotri69]

Thank you very much @kstonekuan for review and merging this pr , now I can solve one more bug