planner

planner

LearningPlanner: converts a diagnosis into a frozen LearningPlan.

Makes a single structured-output teacher call (no tools, no multi-turn) to generate typed edits for each failure cluster. Post-processes the edits with risk tier assignment and patch/replace downgrade.

See spec §6.

Classes

LearningPlanner

LearningPlanner(*, teacher_engine: Any, teacher_model: str, session_id: str, session_dir: Path, prompt_reader: Callable[[str], str])

Converts a diagnosis into a frozen LearningPlan.

PARAMETER	DESCRIPTION
teacher_engine	CloudEngine (or mock) for the planner call. TYPE: Any
teacher_model	Frontier model id. TYPE: str
session_id	Current session id. TYPE: str
session_dir	Path for persisting plan.json and teacher_traces/plan.jsonl. TYPE: Path
prompt_reader	Callable that takes a target string and returns the current prompt content. Used by the patch/replace downgrade logic. TYPE: Callable[[str], str]

Source code in src/openjarvis/learning/distillation/plan/planner.py

def __init__(
    self,
    *,
    teacher_engine: Any,
    teacher_model: str,
    session_id: str,
    session_dir: Path,
    prompt_reader: Callable[[str], str],
) -> None:
    self._engine = teacher_engine
    self._model = teacher_model
    self._session_id = session_id
    self._session_dir = Path(session_dir)
    self._prompt_reader = prompt_reader

Functions

run

run(*, diagnosis_md: str, clusters: list[FailureCluster]) -> LearningPlan

Execute the plan phase.

PARAMETER	DESCRIPTION
diagnosis_md	The teacher's diagnosis markdown from phase 1. TYPE: str
clusters	Failure clusters from phase 1. TYPE: list[FailureCluster]

RETURNS	DESCRIPTION
LearningPlan	The frozen plan, also persisted to plan.json.

Source code in src/openjarvis/learning/distillation/plan/planner.py

def run(
    self,
    *,
    diagnosis_md: str,
    clusters: list[FailureCluster],
) -> LearningPlan:
    """Execute the plan phase.

    Parameters
    ----------
    diagnosis_md :
        The teacher's diagnosis markdown from phase 1.
    clusters :
        Failure clusters from phase 1.

    Returns
    -------
    LearningPlan
        The frozen plan, also persisted to ``plan.json``.
    """
    self._session_dir.mkdir(parents=True, exist_ok=True)

    # Validate clusters — drop those without evidence
    surviving, dropped = _validate_clusters(clusters)

    # Build the user prompt with diagnosis and cluster info
    cluster_json = json.dumps(
        [c.model_dump() for c in surviving],
        indent=2,
        default=str,
    )
    user_prompt = (
        f"## Diagnosis\n\n{diagnosis_md}\n\n"
        f"## Surviving Failure Clusters\n\n```json\n{cluster_json}\n```\n\n"
        "Propose edits for these clusters. Respond with ONLY JSON: "
        '{"edits": [...]}'
    )

    # Make the teacher call
    from openjarvis.core.types import Message, Role

    messages = [
        Message(role=Role.SYSTEM, content=_PLANNER_SYSTEM_PROMPT),
        Message(role=Role.USER, content=user_prompt),
    ]
    result = self._engine.generate(
        messages=messages,
        model=self._model,
        max_tokens=4096,
    )

    cost_usd = result.get("cost_usd", 0.0)
    content = result.get("content", "")

    # Persist teacher trace
    self._persist_trace(content, cost_usd, result)

    # Parse edits from response
    edits = self._parse_edits(content)

    # Post-process: assign tiers deterministically
    edits = assign_tiers(edits)

    # Post-process: downgrade large patches to replacements
    edits = [
        maybe_downgrade_to_replace(e, prompt_reader=self._prompt_reader)
        for e in edits
    ]

    # Wire clusters ↔ edits
    surviving = self._wire_cluster_edit_ids(surviving, edits)

    # Build the plan
    all_clusters = surviving + dropped
    plan = LearningPlan(
        session_id=self._session_id,
        diagnosis_summary=diagnosis_md,
        failure_clusters=all_clusters,
        edits=edits,
        teacher_model=self._model,
        estimated_cost_usd=cost_usd,
        created_at=datetime.now(timezone.utc),
    )

    # Persist plan.json
    plan_path = self._session_dir / "plan.json"
    plan_path.write_text(plan.model_dump_json(indent=2), encoding="utf-8")

    return plan

Functions