solidworks_mcp.ui.routers.checkpoint

solidworks_mcp.ui.routers.checkpoint

Checkpoint execution route for the Prefab CAD dashboard.

Attributes

DEFAULT_SESSION_ID module-attribute

DEFAULT_SESSION_ID = 'prefab-dashboard'

router module-attribute

router = APIRouter()

Classes

CheckpointExecuteRequest

Bases: BaseModel

Request payload for checkpoint execution.

Functions

execute_checkpoint async

execute_checkpoint(payload: CheckpointExecuteRequest) -> dict[str, Any]

Execute the next pending design checkpoint against the SolidWorks adapter.

Source code in src/solidworks_mcp/ui/routers/checkpoint.py

@router.post("/api/ui/checkpoints/execute-next")
async def execute_checkpoint(payload: CheckpointExecuteRequest) -> dict[str, Any]:
    """Execute the next pending design checkpoint against the SolidWorks adapter."""
    return await execute_next_checkpoint(payload.session_id)

execute_next_checkpoint async

execute_next_checkpoint(session_id: str = DEFAULT_SESSION_ID, *, db_path: Path | None = None) -> dict[str, Any]

Find and execute the next un-executed checkpoint in the session plan.

Parameters:

Name	Type	Description	Default
session_id	str	Dashboard session identifier.	DEFAULT_SESSION_ID
db_path	Path | None	Optional SQLite path override.	None

Returns:

Type	Description
dict[str, Any]	Full dashboard state payload.

Source code in src/solidworks_mcp/ui/services/checkpoint_service.py

async def execute_next_checkpoint(
    session_id: str = DEFAULT_SESSION_ID,
    *,
    db_path: Path | None = None,
) -> dict[str, Any]:
    """Find and execute the next un-executed checkpoint in the session plan.

    Args:
        session_id: Dashboard session identifier.
        db_path: Optional SQLite path override.

    Returns:
        Full dashboard state payload.
    """
    from .session_service import build_dashboard_state, ensure_dashboard_session  # noqa: PLC0415

    session_row = ensure_dashboard_session(session_id, db_path=db_path)
    checkpoints = list_plan_checkpoints(session_id, db_path=db_path)
    target = next((row for row in checkpoints if not row["executed"]), None)
    if target is None:
        merge_metadata(
            session_id,
            db_path=db_path,
            latest_message="All checkpoints have already been executed.",
        )
        return build_dashboard_state(session_id, db_path=db_path)

    planned = parse_json_blob(target["planned_action_json"])
    session_meta = parse_json_blob(session_row.get("metadata_json"))

    if _should_open_empty_part(session_meta, planned):
        await _open_empty_part_before_checkpoint(
            session_id=session_id,
            session_row=session_row,
            db_path=db_path,
            api_origin=DEFAULT_API_ORIGIN,
        )

    run_summary = await _run_checkpoint_tools(
        planned,
        session_id=session_id,
        checkpoint_index=int(target["checkpoint_index"]),
        checkpoint_title=str(target["title"]),
        repair_context={
            "active_sketch_plane": session_meta.get("active_sketch_plane"),
            "last_sketch_name": session_meta.get("last_sketch_name"),
            "user_goal": session_row.get("user_goal") or DEFAULT_USER_GOAL,
        },
    )
    failed_tools = run_summary["failed_tools"]
    tool_runs = run_summary["tool_runs"]
    mocked_tools = run_summary.get("mocked_tools", [])
    executed = not failed_tools

    if failed_tools:
        message = (
            f"Checkpoint {target['checkpoint_index']} failed on tools: "
            f"{', '.join(failed_tools)}."
        )
    elif mocked_tools and not any(r.get("status") == "success" for r in tool_runs):
        message = (
            f"Checkpoint {target['checkpoint_index']} executed "
            f"(tools MOCKED: {', '.join(mocked_tools)})."
        )
    else:
        message = (
            f"Executed checkpoint {target['checkpoint_index']}: {target['title']}."
        )

    result_json = json.dumps(
        {
            "status": "success" if executed else "error",
            "message": message,
            "tools": _planned_tools(planned),
            "tool_runs": tool_runs,
            "failed_tools": failed_tools,
            "script_path": run_summary.get("script_path", ""),
            "script_text": run_summary.get("script_text", ""),
            "stdout_tail": str(run_summary.get("stdout_text", ""))[-4000:],
            "stderr_tail": str(run_summary.get("stderr_text", ""))[-4000:],
            "validation_failures": run_summary.get("validation_failures", []),
        },
        ensure_ascii=True,
    )
    update_plan_checkpoint(
        int(target["id"]),
        approved_by_user=True,
        executed=executed,
        result_json=result_json,
        db_path=db_path,
    )

    for tool_run in tool_runs:
        insert_tool_call_record(
            session_id=session_id,
            checkpoint_id=int(target["id"]),
            tool_name=tool_run["tool"],
            input_json=json.dumps(planned, ensure_ascii=True),
            output_json=json.dumps(tool_run, ensure_ascii=True),
            success=tool_run["status"] == "success",
            db_path=db_path,
        )

    upsert_design_session(
        session_id=session_id,
        user_goal=session_row.get("user_goal") or DEFAULT_USER_GOAL,
        source_mode=session_row.get("source_mode") or DEFAULT_SOURCE_MODE,
        accepted_family=session_row.get("accepted_family"),
        status="executing" if executed else "error",
        current_checkpoint_index=(
            target["checkpoint_index"]
            if executed
            else session_row.get("current_checkpoint_index") or 0
        ),
        metadata_json=session_row.get("metadata_json"),
        db_path=db_path,
    )
    merge_metadata(
        session_id,
        db_path=db_path,
        latest_message=message,
        latest_error_text=(message if failed_tools else ""),
        remediation_hint=(
            "Open the generated checkpoint script, repair planned parameters, then rerun this checkpoint."
            if failed_tools
            else ""
        ),
    )
    return build_dashboard_state(session_id, db_path=db_path)