loop

loop

Per-edit execution loop for the distillation execute phase.

Iterates over a plan's edits, handles tier routing, validates, and applies. Does NOT include the benchmark gate — that's wired in M5.

See spec §7.2.

Classes

Functions

execute_edits

execute_edits(*, edits: list[Edit], ctx: ApplyContext, autonomy_mode: AutonomyMode, registry: EditApplierRegistry | None = None) -> list[EditOutcome]

Execute a list of edits, returning outcomes for each.

This loop handles tier routing and validation but does NOT include the benchmark gate (wired in M5).

PARAMETER	DESCRIPTION
edits	The edits to process. TYPE: list[Edit]
ctx	Shared context with config paths. TYPE: ApplyContext
autonomy_mode	How aggressively to apply edits. TYPE: AutonomyMode
registry	Optional pre-built registry. If None, builds the default. TYPE: EditApplierRegistry | None DEFAULT: None

Source code in src/openjarvis/learning/distillation/execute/loop.py

def execute_edits(
    *,
    edits: list[Edit],
    ctx: ApplyContext,
    autonomy_mode: AutonomyMode,
    registry: EditApplierRegistry | None = None,
) -> list[EditOutcome]:
    """Execute a list of edits, returning outcomes for each.

    This loop handles tier routing and validation but does NOT include
    the benchmark gate (wired in M5).

    Parameters
    ----------
    edits :
        The edits to process.
    ctx :
        Shared context with config paths.
    autonomy_mode :
        How aggressively to apply edits.
    registry :
        Optional pre-built registry. If None, builds the default.
    """
    if registry is None:
        registry = _build_registry()

    outcomes: list[EditOutcome] = []

    for edit in edits:
        # Manual mode: everything goes to review
        if autonomy_mode == AutonomyMode.MANUAL:
            outcomes.append(
                EditOutcome(
                    edit_id=edit.id,
                    status="pending_review",
                    benchmark_delta=None,
                    cluster_deltas={},
                    error=None,
                    applied_at=None,
                )
            )
            continue

        # Manual tier: always skip
        if edit.risk_tier == EditRiskTier.MANUAL:
            outcomes.append(
                EditOutcome(
                    edit_id=edit.id,
                    status="skipped",
                    benchmark_delta=None,
                    cluster_deltas={},
                    error="manual tier, requires explicit approval",
                    applied_at=None,
                )
            )
            continue

        # Review tier in tiered mode: route to pending
        if (
            edit.risk_tier == EditRiskTier.REVIEW
            and autonomy_mode == AutonomyMode.TIERED
        ):
            outcomes.append(
                EditOutcome(
                    edit_id=edit.id,
                    status="pending_review",
                    benchmark_delta=None,
                    cluster_deltas={},
                    error=None,
                    applied_at=None,
                )
            )
            continue

        # Check if the op is supported
        if not registry.is_supported(edit.op):
            outcomes.append(
                EditOutcome(
                    edit_id=edit.id,
                    status="skipped",
                    benchmark_delta=None,
                    cluster_deltas={},
                    error=f"op {edit.op.value} not implemented in v1",
                    applied_at=None,
                )
            )
            continue

        # Validate
        applier = registry.get(edit.op)
        validation = applier.validate(edit, ctx)
        if not validation.ok:
            outcomes.append(
                EditOutcome(
                    edit_id=edit.id,
                    status="rejected_by_gate",
                    benchmark_delta=None,
                    cluster_deltas={},
                    error=validation.reason,
                    applied_at=None,
                )
            )
            continue

        # Apply
        try:
            applier.apply(edit, ctx)
            outcomes.append(
                EditOutcome(
                    edit_id=edit.id,
                    status="applied",
                    benchmark_delta=None,
                    cluster_deltas={},
                    error=None,
                    applied_at=datetime.now(timezone.utc),
                )
            )
        except Exception as e:
            logger.warning("Edit %s failed: %s", edit.id, e)
            outcomes.append(
                EditOutcome(
                    edit_id=edit.id,
                    status="rejected_by_gate",
                    benchmark_delta=None,
                    cluster_deltas={},
                    error=str(e),
                    applied_at=None,
                )
            )

    return outcomes