cosmotron_mcp.base

write_plan

write_plan(session_dir: str, steps: list) -> dict

Write the execution plan as a DRAFT plan.json and open the plan-approval gate.

Each step is a dict, e.g. {"agent": "analysis_coder", "action": "compute auto-Cℓ bin 0", "outputs": "results/spectra/cls_bin_00.json"}. The written plan is NOT yet approved: this call opens the plan-approval gate, and every pipeline tool refuses until it resolves. Present the numbered plan to the user in chat as information (not a question), then call authorise_gate(session_dir, "plan-approval", "approve") — the confirmation dialogue that call raises IS the approval. If the user declines the dialogue, ask in chat what to change, revise via the planner, and write_plan again (a rewrite always needs a fresh approval).

After approval the orchestrator dispatches strictly by reading plan.json (via read_plan) and marks each step done (set_plan_step_status) as its artefact appears — it never improvises the next step.

Parameters:

Name	Type	Description	Default
session_dir	str	Session directory.	required
steps	list	Ordered list of step dicts, e.g. {"agent": "analysis_coder", "action": "compute auto-Cℓ bin 0", "outputs": "results/spectra/cls_bin_00.json"}.	required

Returns:

Type	Description
dict	{plan_path, n_steps, approved, gate_id} — approved is already
dict	true (and gate_id None) only when the task declared
dict	skip_gates: true or a gate policy auto-resolved it.

Source code in cosmotron_mcp/server.py

@mcp.tool()
@sync_budget_guard
def write_plan(session_dir: str, steps: list) -> dict:
    """Write the execution plan as a DRAFT plan.json and open the plan-approval gate.

    Each step is a dict, e.g. {"agent": "analysis_coder", "action": "compute
    auto-Cℓ bin 0", "outputs": "results/spectra/cls_bin_00.json"}. The written
    plan is NOT yet approved: this call opens the ``plan-approval`` gate, and
    every pipeline tool refuses until it resolves. Present the numbered plan
    to the user in chat as information (not a question), then call
    ``authorise_gate(session_dir, "plan-approval", "approve")`` — the
    confirmation dialogue that call raises IS the approval. If the user
    declines the dialogue, ask in chat what to change, revise via the
    planner, and write_plan again (a rewrite always needs a fresh approval).

    After approval the orchestrator dispatches strictly by reading plan.json
    (via read_plan) and marks each step done (set_plan_step_status) as its
    artefact appears — it never improvises the next step.

    Args:
        session_dir: Session directory.
        steps: Ordered list of step dicts, e.g. ``{"agent": "analysis_coder",
            "action": "compute auto-Cℓ bin 0", "outputs":
            "results/spectra/cls_bin_00.json"}``.

    Returns:
        ``{plan_path, n_steps, approved, gate_id}`` — ``approved`` is already
        true (and ``gate_id`` None) only when the task declared
        ``skip_gates: true`` or a gate policy auto-resolved it.
    """
    path = _write_plan(session_dir, steps)
    plan = _read_plan(session_dir)
    approved = bool(plan.get("approved", False))
    from cosmotron_mcp.gates import PLAN_GATE_ID
    return {
        "plan_path": path,
        "n_steps": len(plan.get("steps", [])),
        "approved": approved,
        "gate_id": None if approved else PLAN_GATE_ID,
    }

read_plan

read_plan(session_dir: str) -> dict

Read the committed plan.json.

The orchestrator calls this each turn to pick the next status == "pending" step.

Raises:

Type	Description
Exception	If no plan has been committed (the plan gate has not run).

Parameters:

Name	Type	Description	Default
session_dir	str	Session directory.	required

Returns:

Type	Description
dict	{steps: [{n, agent, action, status, ...}], ...}.

Source code in cosmotron_mcp/server.py

@mcp.tool()
@sync_budget_guard
def read_plan(session_dir: str) -> dict:
    """Read the committed plan.json.

    The orchestrator calls this each turn to pick the next ``status == "pending"``
    step.

    Raises:
        Exception: If no plan has been committed (the plan gate has not run).

    Args:
        session_dir: Session directory.

    Returns:
        ``{steps: [{n, agent, action, status, ...}], ...}``.
    """
    return _read_plan(session_dir)

set_plan_step_status

set_plan_step_status(session_dir: str, n: int, status: str = 'done') -> dict

Mark plan step n with status (e.g. "done", "running") in plan.json.

Call after a step's artefact is confirmed in results/artefact_registry.json, then mirror the tick into the opencode TODO list.

Parameters:

Name	Type	Description	Default
session_dir	str	Session directory.	required
n	int	Step number to update.	required
status	str	New status, e.g. "done", "running".	'done'

Returns:

Type	Description
dict	The updated plan dict.

Source code in cosmotron_mcp/server.py

@mcp.tool()
@sync_budget_guard
def set_plan_step_status(session_dir: str, n: int, status: str = "done") -> dict:
    """Mark plan step *n* with *status* (e.g. "done", "running") in plan.json.

    Call after a step's artefact is confirmed in results/artefact_registry.json,
    then mirror the tick into the opencode TODO list.

    Args:
        session_dir: Session directory.
        n: Step number to update.
        status: New status, e.g. ``"done"``, ``"running"``.

    Returns:
        The updated plan dict.
    """
    return _set_plan_step_status(session_dir, n, status)