solidworks_mcp.ui.services

solidworks_mcp.ui.services

Public re-exports for the solidworks_mcp.ui.services package.

Import all public service functions from this package to maintain backward compatibility with code that previously imported directly from solidworks_mcp.ui.service.

Attributes

DEFAULT_API_ORIGIN module-attribute

DEFAULT_API_ORIGIN = getenv('SOLIDWORKS_UI_API_ORIGIN', 'http://127.0.0.1:8766')

DEFAULT_PREVIEW_ORIENTATION module-attribute

DEFAULT_PREVIEW_ORIENTATION = 'current'

DEFAULT_RAG_DIR module-attribute

DEFAULT_RAG_DIR = Path('.solidworks_mcp') / 'rag'

DEFAULT_SESSION_ID module-attribute

DEFAULT_SESSION_ID = 'prefab-dashboard'

DEFAULT_SOURCE_MODE module-attribute

DEFAULT_SOURCE_MODE = 'prompt'

DEFAULT_USER_GOAL module-attribute

DEFAULT_USER_GOAL = 'Design a printable mounting component with documented constraints and fastener strategy.'

DEFAULT_WORKFLOW_MODE module-attribute

DEFAULT_WORKFLOW_MODE = 'unselected'

SUPPORTED_MODEL_UPLOAD_SUFFIXES module-attribute

SUPPORTED_MODEL_UPLOAD_SUFFIXES = frozenset({'.sldprt', '.sldasm', '.slddrw'})

__all__ module-attribute

__all__ = ['DEFAULT_API_ORIGIN', 'DEFAULT_PREVIEW_ORIENTATION', 'DEFAULT_RAG_DIR', 'DEFAULT_SESSION_ID', 'DEFAULT_SOURCE_MODE', 'DEFAULT_USER_GOAL', 'DEFAULT_WORKFLOW_MODE', 'SUPPORTED_MODEL_UPLOAD_SUFFIXES', 'ensure_context_dir', 'ensure_preview_dir', 'ensure_uploaded_model_dir', 'feature_grounding_warning_text', 'feature_target_status', 'filter_docs_text', 'is_url_reference', 'materialize_uploaded_model', 'merge_metadata', 'normalize_feature_targets', 'normalize_workflow_mode', 'parse_json_blob', 'persist_ui_action', 'provider_from_model_name', 'provider_has_credentials', 'read_reference_source', 'read_reference_url', 'safe_context_name', 'sanitize_model_path_text', 'sanitize_preview_viewer_url', 'sanitize_ui_text', 'trace_json', 'trace_session_row', 'trace_tool_records', 'workflow_copy', 'accept_family_choice', 'approve_design_brief', 'build_dashboard_state', 'build_dashboard_trace_payload', 'ensure_dashboard_session', 'load_session_context', 'reconcile_manual_edits', 'save_session_context', 'select_workflow_mode', 'update_session_notes', 'update_ui_preferences', 'ClarificationResponse', 'CheckpointCandidate', 'FamilyInspection', 'inspect_family', 'request_clarifications', 'run_go_orchestration', 'execute_next_checkpoint', 'fetch_docs_context', 'ingest_reference_source', 'connect_target_model', 'open_target_model', 'highlight_feature', 'refresh_preview']

Classes

CheckpointCandidate

Bases: BaseModel

One suggested execution checkpoint.

Attributes:

Name	Type	Description
title	str	Short human-readable label.
allowed_tools	list[str]	MCP tool names allowed at this checkpoint.
rationale	str	Reasoning for including this step.
execution	dict[str, Any]	Optional structured execution hints used by checkpoint runner.

ClarificationResponse

Bases: BaseModel

LLM response for goal clarification.

Attributes:

Name	Type	Description
normalized_brief	str	Concise manufacturing-ready description of the design goal.
questions	list[str]	Follow-up questions that unblock the next modelling step.

FamilyInspection

Bases: BaseModel

LLM response for feature-family classification.

Attributes:

Name	Type	Description
family	str	Detected SolidWorks feature family name.
confidence	Literal['low', 'medium', 'high']	Model confidence level.
evidence	list[str]	Supporting evidence lines.
warnings	list[str]	Contradictory or low-confidence warnings.
checkpoints	list[CheckpointCandidate]	Suggested checkpoint plan for human review.

Functions

accept_family_choice

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

Accept the proposed family classification and advance session status.

Parameters:

Name	Type	Description	Default
session_id	str	Dashboard session identifier.	required
family	str | None	Family name to accept; falls back to proposed_family in metadata.	None
db_path	Path | None	Optional override for the SQLite database path.	None

Returns:

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

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

def accept_family_choice(
    session_id: str,
    family: str | None = None,
    *,
    db_path: Path | None = None,
) -> dict[str, Any]:
    """Accept the proposed family classification and advance session status.

    Args:
        session_id: Dashboard session identifier.
        family: Family name to accept; falls back to ``proposed_family`` in metadata.
        db_path: Optional override for the SQLite database path.

    Returns:
        Full dashboard state payload.
    """
    session_row = ensure_dashboard_session(session_id, db_path=db_path)
    metadata = parse_json_blob(session_row.get("metadata_json"))
    accepted_family = family or metadata.get("proposed_family") or "unknown"
    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=accepted_family,
        status="planned",
        current_checkpoint_index=session_row.get("current_checkpoint_index") or 0,
        metadata_json=session_row.get("metadata_json"),
        db_path=db_path,
    )
    persist_ui_action(
        session_id,
        tool_name="ui.accept_family",
        db_path=db_path,
        metadata_updates={
            "accepted_family": accepted_family,
            "latest_message": f"Family accepted: {accepted_family}.",
            "latest_error_text": "",
            "remediation_hint": "",
        },
        input_payload={"family": accepted_family},
    )
    return build_dashboard_state(session_id, db_path=db_path)

approve_design_brief

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

Persist the accepted design goal and return updated dashboard state.

Parameters:

Name	Type	Description	Default
session_id	str	Dashboard session identifier.	required
user_goal	str	Design goal text approved by the user.	required
db_path	Path | None	Optional override for the SQLite database path.	None

Returns:

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

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

def approve_design_brief(
    session_id: str,
    user_goal: str,
    *,
    db_path: Path | None = None,
) -> dict[str, Any]:
    """Persist the accepted design goal and return updated dashboard state.

    Args:
        session_id: Dashboard session identifier.
        user_goal: Design goal text approved by the user.
        db_path: Optional override for the SQLite database path.

    Returns:
        Full dashboard state payload.
    """
    ensure_dashboard_session(session_id, user_goal=user_goal, db_path=db_path)
    persist_ui_action(
        session_id,
        tool_name="ui.approve_brief",
        db_path=db_path,
        user_goal=user_goal,
        metadata_updates={
            "normalized_brief": user_goal,
            "latest_message": "Brief accepted.",
            "latest_error_text": "",
            "remediation_hint": "",
        },
        input_payload={"user_goal": user_goal},
        output_metadata=True,
    )
    return build_dashboard_state(session_id, db_path=db_path)

build_dashboard_state

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

Assemble the complete dashboard payload consumed by the Prefab UI renderer.

This function is the single read-path for all UI state: it reads the session database, merges every sub-component (checkpoints, evidence, snapshots, preview URLs, provider readiness), and returns the DashboardUIState model as a dict.

Parameters:

Name	Type	Description	Default
session_id	str	Dashboard session identifier.	DEFAULT_SESSION_ID
db_path	Path | None	Optional override for the SQLite database path.	None
api_origin	str	Base URL used to construct preview and viewer URLs.	DEFAULT_API_ORIGIN

Returns:

Type	Description
dict[str, Any]	DashboardUIState model dumped to a plain dict.

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

def build_dashboard_state(
    session_id: str = DEFAULT_SESSION_ID,
    *,
    db_path: Path | None = None,
    api_origin: str = DEFAULT_API_ORIGIN,
) -> dict[str, Any]:
    """Assemble the complete dashboard payload consumed by the Prefab UI renderer.

    This function is the single read-path for all UI state: it reads the session
    database, merges every sub-component (checkpoints, evidence, snapshots, preview
    URLs, provider readiness), and returns the ``DashboardUIState`` model as a dict.

    Args:
        session_id: Dashboard session identifier.
        db_path: Optional override for the SQLite database path.
        api_origin: Base URL used to construct preview and viewer URLs.

    Returns:
        ``DashboardUIState`` model dumped to a plain dict.
    """
    import os  # local import

    session_row = ensure_dashboard_session(session_id, db_path=db_path)
    metadata = parse_json_blob(session_row.get("metadata_json"))
    db_ready = bool(session_row)
    workflow_mode = normalize_workflow_mode(metadata.get("workflow_mode"))
    active_model_path = sanitize_model_path_text(metadata.get("active_model_path"))
    is_new_design_clean = workflow_mode == "new_design" and not active_model_path

    # --- Checkpoints ---
    checkpoints = _build_checkpoint_rows(
        session_id, db_path=db_path, is_new_design_clean=is_new_design_clean
    )
    structured_rendering_enabled = bool(checkpoints)
    checkpoints_text = (
        " | ".join(
            f"{item['step']}. {item['goal']} [{item['status']}] via {item['tools']}"
            for item in checkpoints
        )
        if checkpoints
        else "No checkpoints available yet."
    )

    # --- Evidence rows ---
    evidence_rows = _build_evidence_rows(
        session_id,
        db_path=db_path,
        active_model_path=active_model_path,
        is_new_design_clean=is_new_design_clean,
    )
    evidence_rows_text = (
        " | ".join(
            f"{item['source']}: {item['detail']} (score {item['score']})"
            for item in evidence_rows
        )
        if evidence_rows
        else "No evidence links captured yet."
    )

    # --- Tool history ---
    tool_history = list_tool_call_records(session_id, db_path=db_path)
    latest_tool = tool_history[-1]["tool_name"] if tool_history else "waiting"
    tool_history_text = trace_json(trace_tool_records(tool_history[-20:]))

    # --- Preview URL ---
    import time as _time

    preview_url = ""
    preview_status = "No preview captured yet."
    snapshots = list_model_state_snapshots(session_id, db_path=db_path)
    latest_snapshot_path = snapshots[0].get("screenshot_path") if snapshots else None
    if latest_snapshot_path:
        preview_path = Path(latest_snapshot_path)
        if preview_path.exists():
            ts = int(preview_path.stat().st_mtime)
            preview_url = f"{api_origin}/previews/{preview_path.name}?ts={ts}"
            preview_status = (
                f"Synced from SolidWorks current view. Last file: {preview_path.name}"
            )

    # --- Feature tree ---
    selected_feature_name = str(metadata.get("selected_feature_name") or "")
    feature_tree_items = _build_feature_tree(
        session_id,
        db_path=db_path,
        is_new_design_clean=is_new_design_clean,
        selected_feature_name=selected_feature_name,
    )

    # --- 3D viewer URL ---
    preview_viewer_url = sanitize_preview_viewer_url(
        metadata.get("preview_viewer_url"),
        session_id=session_id,
        api_origin=api_origin,
    )
    if (
        not preview_viewer_url
        and bool(metadata.get("preview_stl_ready"))
        and metadata.get("active_model_path")
    ):
        preview_viewer_url = (
            f"{api_origin}/api/ui/viewer/{session_id}?session_id={session_id}&t=0"
        )

    preview_status = sanitize_ui_text(metadata.get("preview_status"), preview_status)

    # --- Family / clarification ---
    family = (
        session_row.get("accepted_family")
        or metadata.get("proposed_family")
        or "unclassified"
    )
    confidence = metadata.get("family_confidence", "pending")
    evidence_text = (
        " | ".join(metadata.get("family_evidence", [])) or "No family evidence yet."
    )
    warning_text = (
        " | ".join(metadata.get("family_warnings", [])) or "No blocking warnings."
    )
    questions = metadata.get("clarifying_questions", [])
    question_text = (
        "\n".join(f"- {item}" for item in questions)
        if questions
        else "No outstanding clarification questions."
    )

    # --- Model / provider ---
    model_name = sanitize_ui_text(
        metadata.get("model_name"),
        os.getenv("SOLIDWORKS_UI_MODEL", "github:openai/gpt-4.1"),
    )
    model_provider = str(
        metadata.get("model_provider") or provider_from_model_name(model_name)
    )
    model_profile = str(metadata.get("model_profile") or "balanced")

    # --- Active model status ---
    active_model_status = sanitize_ui_text(metadata.get("active_model_status"), "")
    if active_model_path and not active_model_status:
        active_model_status = (
            f"Model path set: {Path(active_model_path).name} (connect pending)."
        )
    if not active_model_path and not active_model_status:
        active_model_status = "No active model connected yet."

    # --- Workflow copy ---
    workflow_label, workflow_guidance_text, flow_header_text = workflow_copy(
        workflow_mode, active_model_path
    )

    # --- Local model ---
    local_endpoint = sanitize_ui_text(
        metadata.get("local_endpoint"),
        os.getenv("SOLIDWORKS_UI_LOCAL_ENDPOINT", "http://127.0.0.1:11434/v1"),
    )

    # --- Readiness ---
    readiness = _compute_readiness(metadata, db_ready=db_ready)

    # --- Context text ---
    active_model_name = Path(active_model_path).name if active_model_path else "<none>"
    preview_views = metadata.get("preview_view_urls") or {}
    model_context_lines = [
        f"Model file: {active_model_name}",
        f"Absolute path: {active_model_path or '<none>'}",
        f"Model type: {str(metadata.get('active_model_type') or '<unknown>')}",
        f"Configuration: {str(metadata.get('active_model_configuration') or '<unknown>')}",
        f"Feature tree rows: {len(feature_tree_items)}",
        f"Selected feature: {selected_feature_name or '<none>'}",
        f"Feature targets: {str(metadata.get('feature_target_text') or '<none>')}",
        f"Preview views captured: {', '.join(sorted(preview_views.keys())) or '<none>'}",
        f"Latest preview status: {preview_status}",
    ]
    model_context_text = "\n".join(model_context_lines)
    context_summary = (
        f"{active_model_name} | {str(metadata.get('active_model_type') or 'unknown')}"
        f" | config {str(metadata.get('active_model_configuration') or '<unknown>')}"
        f" | features {len(feature_tree_items)}"
    )

    fg_warning = feature_grounding_warning_text(
        active_model_path=active_model_path,
        feature_target_text=str(metadata.get("feature_target_text") or ""),
        feature_tree_count=len(feature_tree_items),
    )

    canonical_prompt_text = "\n".join(
        [
            f"Goal: {session_row.get('user_goal') or DEFAULT_USER_GOAL}",
            f"Assumptions: {sanitize_ui_text(metadata.get('assumptions_text'), '') or '<none>'}",
            f"Active model path: {active_model_path or '<none>'}",
            f"Active model status: {active_model_status}",
            f"Feature targets: {str(metadata.get('feature_target_text') or '<none>')}",
            f"Feature target status: {str(metadata.get('feature_target_status') or '<none>')}",
            f"Accepted/proposed family: {session_row.get('accepted_family') or metadata.get('proposed_family') or '<none>'}",
            f"RAG provenance: {str(metadata.get('rag_provenance_text') or '<none>')}",
            f"Docs context: {str(metadata.get('docs_context_text') or '<none>')}",
            f"Engineering notes: {str(metadata.get('notes_text') or '<none>')}",
        ]
    )

    state = DashboardUIState(
        session_id=session_id,
        workflow_mode=workflow_mode,
        workflow_label=workflow_label,
        workflow_guidance_text=workflow_guidance_text,
        user_goal=session_row.get("user_goal") or DEFAULT_USER_GOAL,
        flow_header_text=flow_header_text,
        assumptions_text=sanitize_ui_text(
            metadata.get("assumptions_text"),
            "Assume PETG, 0.4mm nozzle, 0.2mm layers, and 0.30mm mating clearance unless overridden.",
        ),
        active_model_path=active_model_path,
        active_model_status=active_model_status,
        active_model_type=str(metadata.get("active_model_type") or ""),
        active_model_configuration=str(
            metadata.get("active_model_configuration") or ""
        ),
        feature_target_text=str(metadata.get("feature_target_text") or ""),
        feature_target_status=str(
            metadata.get("feature_target_status")
            or "No grounded feature target selected."
        ),
        feature_grounding_warning_text=fg_warning,
        normalized_brief=(
            metadata.get("normalized_brief")
            or session_row.get("user_goal")
            or DEFAULT_USER_GOAL
        ),
        clarifying_questions_text=question_text,
        proposed_family=family,
        family_confidence=confidence,
        family_evidence_text=evidence_text,
        family_warning_text=warning_text,
        accepted_family=session_row.get("accepted_family") or "",
        checkpoints=checkpoints,
        checkpoints_text=checkpoints_text,
        evidence_rows=evidence_rows,
        evidence_rows_text=evidence_rows_text,
        structured_rendering_enabled=structured_rendering_enabled,
        manual_sync_ready=False,
        preview_url=preview_url,
        preview_status=preview_status,
        preview_orientation=metadata.get(
            "preview_orientation", DEFAULT_PREVIEW_ORIENTATION
        ),
        latest_message=metadata.get("latest_message", "Ready."),
        latest_tool=latest_tool,
        latest_error_text=str(metadata.get("latest_error_text") or ""),
        remediation_hint=str(metadata.get("remediation_hint") or ""),
        model_provider=model_provider,
        model_name=model_name,
        model_profile=model_profile,
        local_endpoint=local_endpoint,
        local_model_status_text=str(
            metadata.get("local_model_status_text") or "Local model controls idle."
        ),
        local_model_busy=bool(metadata.get("local_model_busy") or False),
        local_model_available=bool(metadata.get("local_model_available") or False),
        local_model_recommended_tier=str(
            metadata.get("local_model_recommended_tier") or ""
        ),
        local_model_recommended_ollama_model=str(
            metadata.get("local_model_recommended_ollama_model") or ""
        ),
        local_model_pull_command=str(metadata.get("local_model_pull_command") or ""),
        local_model_label=str(metadata.get("local_model_label") or ""),
        rag_source_path=str(metadata.get("rag_source_path") or ""),
        rag_namespace=str(metadata.get("rag_namespace") or "engineering-reference"),
        rag_status=str(
            metadata.get("rag_status") or "No retrieval source ingested yet."
        ),
        rag_index_path=str(metadata.get("rag_index_path") or ""),
        rag_chunk_count=int(metadata.get("rag_chunk_count") or 0),
        rag_provenance_text=str(
            metadata.get("rag_provenance_text")
            or "No retrieval provenance available yet."
        ),
        docs_query=str(metadata.get("docs_query") or "SolidWorks MCP endpoints"),
        docs_context_text=str(
            metadata.get("docs_context_text") or "No docs context loaded yet."
        ),
        notes_text=str(metadata.get("notes_text") or ""),
        orchestration_status=str(metadata.get("orchestration_status") or "Ready."),
        context_save_status=str(metadata.get("context_save_status") or ""),
        context_load_status=str(metadata.get("context_load_status") or ""),
        context_name_input=str(metadata.get("context_name_input") or session_id),
        context_file_input=str(metadata.get("last_context_file") or ""),
        readiness_provider_configured=readiness["readiness_provider_configured"],
        readiness_adapter_mode=readiness["readiness_adapter_mode"],
        readiness_preview_ready=readiness["readiness_preview_ready"],
        readiness_db_ready=readiness["readiness_db_ready"],
        readiness_summary=readiness["readiness_summary"],
        context_used_pct=38,
        context_text=context_summary,
        model_context_text=model_context_text,
        canonical_prompt_text=canonical_prompt_text,
        tool_history_text=tool_history_text,
        api_origin=api_origin,
        preview_viewer_url=preview_viewer_url,
        preview_view_urls=metadata.get("preview_view_urls") or {},
        user_clarification_answer=str(metadata.get("user_clarification_answer") or ""),
        mocked_tools_text=(
            "MOCKED tools: " + ", ".join(metadata.get("mocked_tools", []))
            if metadata.get("mocked_tools")
            else ""
        ),
        feature_tree_items=feature_tree_items,
        selected_feature_name=str(metadata.get("selected_feature_name") or ""),
    ).model_dump()

    logger.debug(
        "[ui.trace.state] session_id={} model_path={} selected={} feature_rows={} preview_views={} latest_tool={}",
        session_id,
        state.get("active_model_path") or "<none>",
        state.get("selected_feature_name") or "<none>",
        len(state.get("feature_tree_items") or []),
        list((state.get("preview_view_urls") or {}).keys()),
        state.get("latest_tool") or "waiting",
    )
    return state

build_dashboard_trace_payload

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

Assemble the verbose debug/trace payload for the operator trace panel.

Includes the raw session row, full metadata, complete tool-call history, and the assembled DashboardUIState — all serialised to both Python dicts and pretty-printed JSON strings for easy inspection in the UI.

Parameters:

Name	Type	Description	Default
session_id	str	Dashboard session identifier.	DEFAULT_SESSION_ID
db_path	Path | None	Optional override for the SQLite database path.	None
api_origin	str	API origin used for URL generation.	DEFAULT_API_ORIGIN

Returns:

Type	Description
dict[str, Any]	Dict with session_row, metadata, state, and tool_records sections.

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

def build_dashboard_trace_payload(
    session_id: str = DEFAULT_SESSION_ID,
    *,
    db_path: Path | None = None,
    api_origin: str = DEFAULT_API_ORIGIN,
) -> dict[str, Any]:
    """Assemble the verbose debug/trace payload for the operator trace panel.

    Includes the raw session row, full metadata, complete tool-call history, and the
    assembled ``DashboardUIState`` — all serialised to both Python dicts and
    pretty-printed JSON strings for easy inspection in the UI.

    Args:
        session_id: Dashboard session identifier.
        db_path: Optional override for the SQLite database path.
        api_origin: API origin used for URL generation.

    Returns:
        Dict with ``session_row``, ``metadata``, ``state``, and ``tool_records`` sections.
    """
    ensure_dashboard_session(session_id, db_path=db_path)
    session_row = get_design_session(session_id, db_path=db_path) or {}
    metadata = parse_json_blob(session_row.get("metadata_json"))
    state = build_dashboard_state(session_id, db_path=db_path, api_origin=api_origin)
    tool_records = trace_tool_records(
        list_tool_call_records(session_id, db_path=db_path)
    )
    session_row_payload = trace_session_row(session_row)

    payload = {
        "session_id": session_id,
        "workflow_mode": state.get("workflow_mode", DEFAULT_WORKFLOW_MODE),
        "latest_message": state.get("latest_message", "Ready."),
        "latest_error_text": state.get("latest_error_text", ""),
        "debug_summary": (
            f"workflow={state.get('workflow_mode', DEFAULT_WORKFLOW_MODE)}"
            f" | model_path={state.get('active_model_path', '') or '<none>'}"
            f" | latest_tool={state.get('latest_tool', 'waiting')}"
            f" | tool_records={len(tool_records)}"
        ),
        "session_row": session_row_payload,
        "session_row_text": trace_json(session_row_payload),
        "metadata": metadata,
        "metadata_text": trace_json(metadata),
        "state": state,
        "state_text": trace_json(state),
        "tool_records": tool_records,
        "tool_records_text": trace_json(tool_records),
    }
    logger.debug(
        "[ui.trace.snapshot] session_id={} model_path={} selected={} tool_records={} preview_views={}",
        session_id,
        state.get("active_model_path") or "<none>",
        state.get("selected_feature_name") or "<none>",
        len(tool_records),
        list((state.get("preview_view_urls") or {}).keys()),
    )
    return payload

connect_target_model async

connect_target_model(session_id: str, *, model_path: str | None = None, uploaded_files: list[dict[str, Any]] | None = None, feature_target_text: str | None = None, db_path: Path | None = None, api_origin: str = DEFAULT_API_ORIGIN) -> dict[str, Any]

Open a target model, inspect its feature tree, and trigger a preview refresh.

Parameters:

Name	Type	Description	Default
session_id	str	Dashboard session identifier.	required
model_path	str | None	Absolute path to the SolidWorks model file.	None
uploaded_files	list[dict[str, Any]] | None	List of base64-encoded file upload dicts.	None
feature_target_text	str | None	Optional comma-separated feature target references.	None
db_path	Path | None	Optional SQLite path override.	None
api_origin	str	Base URL of the running FastAPI server.	DEFAULT_API_ORIGIN

Returns:

Type	Description
dict[str, Any]	Full dashboard state payload (includes preview state after the refresh).

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

async def connect_target_model(
    session_id: str,
    *,
    model_path: str | None = None,
    uploaded_files: list[dict[str, Any]] | None = None,
    feature_target_text: str | None = None,
    db_path: Path | None = None,
    api_origin: str = DEFAULT_API_ORIGIN,
) -> dict[str, Any]:
    """Open a target model, inspect its feature tree, and trigger a preview refresh.

    Args:
        session_id: Dashboard session identifier.
        model_path: Absolute path to the SolidWorks model file.
        uploaded_files: List of base64-encoded file upload dicts.
        feature_target_text: Optional comma-separated feature target references.
        db_path: Optional SQLite path override.
        api_origin: Base URL of the running FastAPI server.

    Returns:
        Full dashboard state payload (includes preview state after the refresh).
    """
    from .session_service import build_dashboard_state, ensure_dashboard_session  # noqa: PLC0415
    from .preview_service import refresh_preview  # noqa: PLC0415

    ensure_dashboard_session(session_id, db_path=db_path)
    adapter = None
    resolved_path: Path | None = None

    logger.info(
        "[ui.connect_target_model] session_id={} model_path={} uploaded_files={} feature_targets={}",
        session_id,
        model_path,
        len(uploaded_files) if uploaded_files else 0,
        feature_target_text or "",
    )

    resolved_path = _resolve_model_path(
        session_id,
        model_path=model_path,
        uploaded_files=uploaded_files,
        feature_target_text=feature_target_text,
        db_path=db_path,
        api_origin=api_origin,
    )
    if resolved_path is None:
        return build_dashboard_state(session_id, db_path=db_path, api_origin=api_origin)

    config = load_config()
    adapter = await create_adapter(config)
    model_info: dict[str, Any] = {}
    features: list[dict[str, Any]] = []
    attach_succeeded = False
    tool_input = {
        "model_path": str(resolved_path.resolve()),
        "uploaded_file_name": uploaded_files[0].get("name") if uploaded_files else None,
        "feature_target_text": feature_target_text or "",
    }
    try:
        await adapter.connect()
        open_result = await adapter.open_model(str(resolved_path.resolve()))
        if not open_result.is_success:
            raise RuntimeError(open_result.error or "Failed to open target model.")

        if hasattr(adapter, "get_model_info"):
            info_result = await adapter.get_model_info()
            if info_result.is_success and isinstance(info_result.data, dict):
                model_info = info_result.data

        if hasattr(adapter, "list_features"):
            feature_result = await adapter.list_features(include_suppressed=True)
            if feature_result.is_success and isinstance(feature_result.data, list):
                features = [
                    item for item in feature_result.data if isinstance(item, dict)
                ]

        from ._utils import feature_target_status  # noqa: PLC0415

        classification = classify_feature_tree_snapshot(model_info, features)
        target_status, matched_targets, missing_targets = feature_target_status(
            features, feature_target_text
        )

        snapshot_id = insert_model_state_snapshot(
            session_id=session_id,
            model_path=str(resolved_path.resolve()),
            feature_tree_json=json.dumps(features, ensure_ascii=True),
            state_fingerprint=f"{resolved_path.resolve()}::{resolved_path.stat().st_mtime_ns}",
            db_path=db_path,
        )
        for evidence_line in classification.get("evidence", []):
            insert_evidence_link(
                session_id=session_id,
                source_type="active_model",
                source_id=str(resolved_path.resolve()),
                relevance_score=0.96,
                rationale=str(evidence_line),
                payload_json=json.dumps(classification, ensure_ascii=True),
                db_path=db_path,
            )

        if matched_targets or missing_targets:
            insert_evidence_link(
                session_id=session_id,
                source_type="feature_target",
                source_id=str(resolved_path.resolve()),
                relevance_score=0.9 if matched_targets else 0.4,
                rationale=target_status,
                payload_json=json.dumps(
                    {"matched": matched_targets, "missing": missing_targets},
                    ensure_ascii=True,
                ),
                db_path=db_path,
            )

        metadata = merge_metadata(
            session_id,
            db_path=db_path,
            workflow_mode="edit_existing",
            active_model_path=str(resolved_path.resolve()),
            active_model_status=(
                f"Attached model: {resolved_path.name}"
                f" | type={model_info.get('type', 'unknown')}"
                f" | features={len(features)}"
            ),
            active_model_type=str(model_info.get("type") or ""),
            active_model_configuration=str(
                model_info.get("configuration") or "Default"
            ),
            feature_target_text=feature_target_text or "",
            feature_target_status=target_status,
            proposed_family=classification.get("family") or "unknown",
            family_confidence=classification.get("confidence") or "low",
            family_evidence=classification.get("evidence") or [],
            family_warnings=classification.get("warnings") or [],
            latest_message=(
                f"Opened target model {resolved_path.name}. Generating preview views..."
            ),
            preview_status="Generating preview views from attached model...",
            preview_stl_ready=False,
            preview_png_ready=False,
            preview_viewer_url="",
            latest_error_text="",
            remediation_hint="",
            latest_snapshot_id=snapshot_id,
        )
        insert_tool_call_record(
            session_id=session_id,
            tool_name="ui.connect_target_model",
            input_json=json.dumps(tool_input, ensure_ascii=True),
            output_json=json.dumps(metadata, ensure_ascii=True),
            success=True,
            db_path=db_path,
        )
        attach_succeeded = True
    except Exception as exc:
        logger.exception(
            "[ui.connect_target_model] failed session_id={} path={} error={}",
            session_id,
            str(resolved_path.resolve()) if resolved_path else "<none>",
            exc,
        )
        merge_metadata(
            session_id,
            db_path=db_path,
            workflow_mode="edit_existing",
            active_model_path=str(resolved_path.resolve()),
            feature_target_text=feature_target_text or "",
            latest_message="Failed to attach target model.",
            preview_status="Preview generation failed while attaching model.",
            latest_error_text=str(exc),
            remediation_hint="Open SolidWorks, verify COM access, and retry with a valid .sldprt/.sldasm path.",
        )
        insert_tool_call_record(
            session_id=session_id,
            tool_name="ui.connect_target_model",
            input_json=json.dumps(tool_input, ensure_ascii=True),
            output_json=json.dumps({"error": str(exc)}, ensure_ascii=True),
            success=False,
            db_path=db_path,
        )

    if attach_succeeded:
        try:
            return await refresh_preview(
                session_id,
                orientation=DEFAULT_PREVIEW_ORIENTATION,
                db_path=db_path,
                preview_dir=ensure_preview_dir(),
                api_origin=api_origin,
                adapter_override=adapter,
                active_model_path_override=str(resolved_path.resolve()),
                reopen_active_model=False,
            )
        except Exception as refresh_exc:
            logger.warning(
                "[ui.connect_target_model] post-attach preview refresh failed: {}",
                str(refresh_exc),
            )

    if adapter is not None:
        try:
            await adapter.disconnect()
        except Exception:
            logger.debug("Adapter disconnect failed during target-model cleanup")

    return build_dashboard_state(session_id, db_path=db_path, api_origin=api_origin)

ensure_context_dir

ensure_context_dir(context_dir: Path | None = None) -> Path

Create and return the dashboard context snapshot directory.

Parameters:

Name	Type	Description	Default
context_dir	Path | None	Override directory; defaults to .solidworks_mcp/ui_context.	None

Returns:

Type	Description
Path	Resolved Path that is guaranteed to exist.

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

def ensure_context_dir(context_dir: Path | None = None) -> Path:
    """Create and return the dashboard context snapshot directory.

    Args:
        context_dir: Override directory; defaults to ``.solidworks_mcp/ui_context``.

    Returns:
        Resolved ``Path`` that is guaranteed to exist.
    """
    resolved = context_dir or _DEFAULT_CONTEXT_DIR
    resolved.mkdir(parents=True, exist_ok=True)
    return resolved

ensure_dashboard_session

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

Ensure one dashboard session row and default checkpoints exist.

Creates the session row on first call and inserts default checkpoints when none are present yet. On subsequent calls it updates the user_goal column when the supplied goal differs from the stored one.

Parameters:

Name	Type	Description	Default
session_id	str	Dashboard session identifier.	DEFAULT_SESSION_ID
user_goal	str | None	Optional initial or updated design goal.	None
db_path	Path | None	Optional override for the SQLite database path.	None

Returns:

Type	Description
dict[str, Any]	The current session row dict.

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

def ensure_dashboard_session(
    session_id: str = DEFAULT_SESSION_ID,
    *,
    user_goal: str | None = None,
    db_path: Path | None = None,
) -> dict[str, Any]:
    """Ensure one dashboard session row and default checkpoints exist.

    Creates the session row on first call and inserts default checkpoints when
    none are present yet. On subsequent calls it updates the ``user_goal`` column
    when the supplied goal differs from the stored one.

    Args:
        session_id: Dashboard session identifier.
        user_goal: Optional initial or updated design goal.
        db_path: Optional override for the SQLite database path.

    Returns:
        The current session row dict.
    """
    session_row = get_design_session(session_id, db_path=db_path)
    requested_goal = sanitize_ui_text(user_goal, "") if user_goal is not None else ""
    if session_row is None:
        upsert_design_session(
            session_id=session_id,
            user_goal=requested_goal or DEFAULT_USER_GOAL,
            source_mode=DEFAULT_SOURCE_MODE,
            status="inspect",
            metadata_json=json.dumps(
                {
                    "normalized_brief": requested_goal or DEFAULT_USER_GOAL,
                    "preview_orientation": DEFAULT_PREVIEW_ORIENTATION,
                },
                ensure_ascii=True,
            ),
            db_path=db_path,
        )
    elif requested_goal and requested_goal != session_row["user_goal"]:
        upsert_design_session(
            session_id=session_id,
            user_goal=requested_goal,
            source_mode=session_row["source_mode"],
            accepted_family=session_row["accepted_family"],
            status=session_row["status"],
            current_checkpoint_index=session_row["current_checkpoint_index"],
            metadata_json=session_row["metadata_json"],
            db_path=db_path,
        )

    checkpoints = list_plan_checkpoints(session_id, db_path=db_path)
    if not checkpoints:
        base_goal = requested_goal or session_row.get("user_goal") if session_row else ""
        for index, spec in enumerate(_checkpoint_specs_for_goal(str(base_goal)), start=1):
            insert_plan_checkpoint(
                session_id=session_id,
                checkpoint_index=index,
                title=spec["title"],
                planned_action_json=json.dumps(spec, ensure_ascii=True),
                approved_by_user=index == 1,
                db_path=db_path,
            )

    return get_design_session(session_id, db_path=db_path) or {}

ensure_preview_dir

ensure_preview_dir(preview_dir: Path | None = None) -> Path

Create and return the preview image directory.

Parameters:

Name	Type	Description	Default
preview_dir	Path | None	Override directory; defaults to .solidworks_mcp/ui_previews.	None

Returns:

Type	Description
Path	Resolved Path that is guaranteed to exist.

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

def ensure_preview_dir(preview_dir: Path | None = None) -> Path:
    """Create and return the preview image directory.

    Args:
        preview_dir: Override directory; defaults to ``.solidworks_mcp/ui_previews``.

    Returns:
        Resolved ``Path`` that is guaranteed to exist.
    """
    resolved = preview_dir or _DEFAULT_PREVIEW_DIR
    resolved.mkdir(parents=True, exist_ok=True)
    return resolved

ensure_uploaded_model_dir

ensure_uploaded_model_dir(upload_dir: Path | None = None) -> Path

Create and return the uploaded-model staging directory.

Parameters:

Name	Type	Description	Default
upload_dir	Path | None	Override directory; defaults to .solidworks_mcp/ui_uploads.	None

Returns:

Type	Description
Path	Resolved Path that is guaranteed to exist.

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

def ensure_uploaded_model_dir(upload_dir: Path | None = None) -> Path:
    """Create and return the uploaded-model staging directory.

    Args:
        upload_dir: Override directory; defaults to ``.solidworks_mcp/ui_uploads``.

    Returns:
        Resolved ``Path`` that is guaranteed to exist.
    """
    resolved = upload_dir or _DEFAULT_UPLOADED_MODEL_DIR
    resolved.mkdir(parents=True, exist_ok=True)
    return resolved

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)

feature_grounding_warning_text

feature_grounding_warning_text(*, active_model_path: str, feature_target_text: str, feature_tree_count: int) -> str

Return a warning string when feature grounding cannot proceed.

Parameters:

Name	Type	Description	Default
active_model_path	str	Currently attached model path.	required
feature_target_text	str	Raw feature target refs from UI.	required
feature_tree_count	int	Number of rows in the current feature tree snapshot.	required

Returns:

Type	Description
str	Warning message, or "" if grounding is possible.

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

def feature_grounding_warning_text(
    *,
    active_model_path: str,
    feature_target_text: str,
    feature_tree_count: int,
) -> str:
    """Return a warning string when feature grounding cannot proceed.

    Args:
        active_model_path: Currently attached model path.
        feature_target_text: Raw feature target refs from UI.
        feature_tree_count: Number of rows in the current feature tree snapshot.

    Returns:
        Warning message, or ``""`` if grounding is possible.
    """
    if not active_model_path:
        return ""
    if not str(feature_target_text or "").strip():
        return ""
    if feature_tree_count > 0:
        return ""
    return (
        "Grounding is unavailable for the current attached model context because "
        "no feature tree rows were returned. Feature refs such as @Boss-Extrude1 "
        "cannot be resolved until the adapter can read the active model tree."
    )

feature_target_status

feature_target_status(features: list[dict[str, Any]], feature_target_text: str | None) -> tuple[str, list[str], list[str]]

Compute matched / missing feature target status.

Parameters:

Name	Type	Description	Default
features	list[dict[str, Any]]	Feature tree rows from the active snapshot.	required
feature_target_text	str | None	Raw feature target input from the UI.	required

Returns:

Type	Description
tuple[str, list[str], list[str]]	Three-tuple of (status_message, matched_names, missing_names).

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

def feature_target_status(
    features: list[dict[str, Any]], feature_target_text: str | None
) -> tuple[str, list[str], list[str]]:
    """Compute matched / missing feature target status.

    Args:
        features: Feature tree rows from the active snapshot.
        feature_target_text: Raw feature target input from the UI.

    Returns:
        Three-tuple of (status_message, matched_names, missing_names).
    """
    requested = normalize_feature_targets(feature_target_text)
    if not requested:
        if str(feature_target_text or "").strip():
            return (
                "No valid feature targets found. Use feature names such as @Boss-Extrude1 or @Sketch2. File paths are ignored.",
                [],
                [],
            )
        return ("No grounded feature target selected.", [], [])

    available = {
        str(feature.get("name") or "").strip().lower(): str(feature.get("name") or "")
        for feature in features
        if str(feature.get("name") or "").strip()
    }
    matched: list[str] = []
    missing: list[str] = []
    for target in requested:
        hit = available.get(target.lower())
        if hit:
            matched.append(hit)
        else:
            missing.append(target)

    if matched and not missing:
        return (
            "Grounded feature target(s): " + ", ".join(f"@{name}" for name in matched),
            matched,
            missing,
        )
    if matched:
        return (
            "Partially grounded target(s): "
            + ", ".join(f"@{name}" for name in matched)
            + " | Missing: "
            + ", ".join(f"@{name}" for name in missing),
            matched,
            missing,
        )
    return (
        "No matching feature targets found for: "
        + ", ".join(f"@{name}" for name in missing),
        matched,
        missing,
    )

fetch_docs_context

fetch_docs_context(session_id: str, *, docs_query: str = '', db_path: Path | None = None, api_origin: str = DEFAULT_API_ORIGIN) -> dict[str, Any]

Fetch docs text from the /docs endpoint and store a filtered context snippet.

Parameters:

Name	Type	Description	Default
session_id	str	Dashboard session identifier.	required
docs_query	str	Keyword(s) to filter the docs text by.	''
db_path	Path | None	Optional override for the SQLite database path.	None
api_origin	str	Base URL of the running FastAPI server.	DEFAULT_API_ORIGIN

Returns:

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

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

def fetch_docs_context(
    session_id: str,
    *,
    docs_query: str = "",
    db_path: Path | None = None,
    api_origin: str = DEFAULT_API_ORIGIN,
) -> dict[str, Any]:
    """Fetch docs text from the ``/docs`` endpoint and store a filtered context snippet.

    Args:
        session_id: Dashboard session identifier.
        docs_query: Keyword(s) to filter the docs text by.
        db_path: Optional override for the SQLite database path.
        api_origin: Base URL of the running FastAPI server.

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

    ensure_dashboard_session(session_id, db_path=db_path)
    docs_url = f"{api_origin}/docs"
    query_text = sanitize_ui_text(docs_query, "solidworks workflow")
    try:
        request = Request(docs_url, headers={"User-Agent": "solidworks-mcp-ui/1.0"})
        with urlopen(request, timeout=8) as response:
            html = response.read().decode("utf-8", errors="ignore")
        extractor = HTMLTextExtractor()
        extractor.feed(html)
        snippet = filter_docs_text(extractor.text(), query_text)
        persist_ui_action(
            session_id,
            tool_name="ui.docs.fetch",
            db_path=db_path,
            metadata_updates={
                "docs_query": query_text,
                "docs_context_text": snippet,
                "latest_message": "Docs context updated from MCP docs endpoint.",
                "latest_error_text": "",
                "remediation_hint": "",
            },
            input_payload={"query": query_text, "url": docs_url},
            output_payload={"chars": len(snippet)},
        )
    except Exception as exc:
        logger.exception("[ui.fetch_docs_context] failed session_id={}", session_id)
        merge_metadata(
            session_id,
            db_path=db_path,
            docs_query=query_text,
            docs_context_text="",
            latest_error_text=str(exc),
            remediation_hint="Verify the /docs endpoint is reachable, then retry docs refresh.",
        )
    return build_dashboard_state(session_id, db_path=db_path, api_origin=api_origin)

filter_docs_text

filter_docs_text(text: str, query: str, *, max_chars: int = 4000) -> str

Filter plain-text docs content to lines most relevant to query.

Returns up to max_chars characters of the most relevant lines, scored by how many query tokens they contain.

Parameters:

Name	Type	Description	Default
text	str	Full plain-text docs content.	required
query	str	Space-separated keyword query.	required
max_chars	int	Maximum characters to return.	4000

Returns:

Type	Description
str	Filtered and truncated text snippet.

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

def filter_docs_text(text: str, query: str, *, max_chars: int = 4000) -> str:
    """Filter plain-text docs content to lines most relevant to *query*.

    Returns up to ``max_chars`` characters of the most relevant lines, scored
    by how many query tokens they contain.

    Args:
        text: Full plain-text docs content.
        query: Space-separated keyword query.
        max_chars: Maximum characters to return.

    Returns:
        Filtered and truncated text snippet.
    """
    if not text:
        return ""
    tokens = {t.lower() for t in query.split() if t}
    lines = text.splitlines()
    scored: list[tuple[int, str]] = []
    for line in lines:
        lower = line.lower()
        score = sum(1 for t in tokens if t in lower)
        if score > 0:
            scored.append((score, line))
    scored.sort(key=lambda x: x[0], reverse=True)
    selected = [line for _, line in scored]
    combined = "\n".join(selected)
    return combined[:max_chars]

highlight_feature async

highlight_feature(session_id: str, feature_name: str, *, db_path: Path | None = None, api_origin: str = DEFAULT_API_ORIGIN) -> dict[str, Any]

Select and highlight a named feature in the active SolidWorks model.

Parameters:

Name	Type	Description	Default
session_id	str	Dashboard session identifier.	required
feature_name	str	Name of the feature to select (must match the feature-tree entry).	required
db_path	Path | None	Optional SQLite path override.	None
api_origin	str	Base URL of the running FastAPI server.	DEFAULT_API_ORIGIN

Returns:

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

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

async def highlight_feature(
    session_id: str,
    feature_name: str,
    *,
    db_path: Path | None = None,
    api_origin: str = DEFAULT_API_ORIGIN,
) -> dict[str, Any]:
    """Select and highlight a named feature in the active SolidWorks model.

    Args:
        session_id: Dashboard session identifier.
        feature_name: Name of the feature to select (must match the feature-tree entry).
        db_path: Optional SQLite path override.
        api_origin: Base URL of the running FastAPI server.

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

    ensure_dashboard_session(session_id, db_path=db_path)
    session_row = get_design_session(session_id, db_path=db_path) or {}
    metadata = parse_json_blob(session_row.get("metadata_json"))
    active_model_path = metadata.get("active_model_path")
    resolved_name = (feature_name or "").strip()
    if not resolved_name:
        merge_metadata(
            session_id,
            db_path=db_path,
            latest_error_text="No feature name provided for selection.",
            remediation_hint="Pass a non-empty feature_name.",
        )
        return build_dashboard_state(session_id, db_path=db_path, api_origin=api_origin)

    try:
        known_feature_names: set[str] = set()
        for snapshot in list_model_state_snapshots(session_id, db_path=db_path):
            raw_tree = snapshot.get("feature_tree_json")
            if not raw_tree:
                continue
            try:
                parsed_tree = json.loads(raw_tree)
            except Exception:
                continue
            if isinstance(parsed_tree, list):
                known_feature_names.update(
                    str(item.get("name") or "").strip()
                    for item in parsed_tree
                    if str(item.get("name") or "").strip()
                )
                if known_feature_names:
                    break

        config = load_config()
        adapter = await create_adapter(config)
        await adapter.connect()
        if active_model_path and hasattr(adapter, "open_model"):
            candidate = Path(str(active_model_path))
            if candidate.exists():
                await adapter.open_model(str(candidate.resolve()))
        selected = False
        entity_type = ""
        selected_name = resolved_name
        if hasattr(adapter, "select_feature"):
            result = await adapter.select_feature(resolved_name)
            if result.is_success and isinstance(result.data, dict):
                selected = bool(result.data.get("selected"))
                entity_type = str(result.data.get("entity_type") or "")
                selected_name = str(result.data.get("selected_name") or resolved_name)
        await adapter.disconnect()
        tracked_only = (not selected) and (resolved_name in known_feature_names)
        merge_metadata(
            session_id,
            db_path=db_path,
            selected_feature_name=resolved_name,
            selected_feature_selector_name=selected_name,
            latest_message=(
                f"Selected '{resolved_name}' ({entity_type}) in SolidWorks."
                if selected
                else (
                    f"Tracking '{resolved_name}' from the feature tree. "
                    "SolidWorks did not expose a direct selectable handle for that row."
                    if tracked_only
                    else f"Could not select feature '{resolved_name}' — name may not match the feature tree."
                )
            ),
            latest_error_text=(
                ""
                if (selected or tracked_only)
                else f"SelectByID2 returned False for '{resolved_name}'."
            ),
            remediation_hint=(
                ""
                if (selected or tracked_only)
                else "Check that the feature name exactly matches the SolidWorks feature tree entry."
            ),
        )
        insert_tool_call_record(
            session_id=session_id,
            tool_name="ui.highlight_feature",
            input_json=json.dumps({"feature_name": resolved_name}, ensure_ascii=True),
            output_json=json.dumps(
                {
                    "selected": selected,
                    "tracked_only": tracked_only,
                    "entity_type": entity_type,
                },
                ensure_ascii=True,
            ),
            success=(selected or tracked_only),
            db_path=db_path,
        )
    except Exception as exc:
        logger.exception("[ui.highlight_feature] failed: {}", exc)
        merge_metadata(
            session_id,
            db_path=db_path,
            latest_error_text=str(exc),
            remediation_hint="Ensure SolidWorks is open with the target model loaded.",
        )
    return build_dashboard_state(session_id, db_path=db_path, api_origin=api_origin)

ingest_reference_source

ingest_reference_source(session_id: str, *, source_path: str, namespace: str, chunk_size: int = 1200, overlap: int = 200, db_path: Path | None = None) -> dict[str, Any]

Ingest a local file or URL into the simple local retrieval index.

Parameters:

Name	Type	Description	Default
session_id	str	Dashboard session identifier.	required
source_path	str	Absolute file path or http/https URL.	required
namespace	str	Namespace key that isolates this index from others.	required
chunk_size	int	Maximum characters per chunk.	1200
overlap	int	Overlapping characters between adjacent chunks.	200
db_path	Path | None	Optional override for the SQLite database path.	None

Returns:

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

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

def ingest_reference_source(
    session_id: str,
    *,
    source_path: str,
    namespace: str,
    chunk_size: int = 1200,
    overlap: int = 200,
    db_path: Path | None = None,
) -> dict[str, Any]:
    """Ingest a local file or URL into the simple local retrieval index.

    Args:
        session_id: Dashboard session identifier.
        source_path: Absolute file path or http/https URL.
        namespace: Namespace key that isolates this index from others.
        chunk_size: Maximum characters per chunk.
        overlap: Overlapping characters between adjacent chunks.
        db_path: Optional override for the SQLite database path.

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

    ensure_dashboard_session(session_id, db_path=db_path)
    source_reference = (source_path or "").strip()
    resolved_namespace = (
        namespace or "engineering-reference"
    ).strip() or "engineering-reference"

    try:
        if is_url_reference(source_reference):
            source_identifier = source_reference
            source_text, source_label = read_reference_url(source_reference)
        else:
            resolved_source = Path(source_reference).expanduser()
            if not resolved_source.exists():
                merge_metadata(
                    session_id,
                    db_path=db_path,
                    rag_source_path=str(resolved_source),
                    rag_namespace=resolved_namespace,
                    rag_status="Reference source path was not found.",
                    latest_error_text=f"Missing reference source: {resolved_source}",
                    remediation_hint=(
                        "Provide an absolute path or an http/https URL for a PDF, "
                        "markdown, text, or HTML source."
                    ),
                )
                return build_dashboard_state(session_id, db_path=db_path)
            source_identifier = str(resolved_source.resolve())
            source_label = resolved_source.name
            source_text = read_reference_source(resolved_source)

        chunks = _chunk_text(source_text, chunk_size=chunk_size, overlap=overlap)
        output_path = DEFAULT_RAG_DIR / f"{resolved_namespace}.json"
        output_path.parent.mkdir(parents=True, exist_ok=True)
        payload = {
            "version": "1.0",
            "namespace": resolved_namespace,
            "source_location": source_identifier,
            "chunk_count": len(chunks),
            "chunks": [
                {
                    "id": f"{resolved_namespace}-{index}",
                    "source": source_identifier,
                    "text": chunk,
                }
                for index, chunk in enumerate(chunks, start=1)
            ],
        }
        output_path.write_text(
            json.dumps(payload, indent=2, ensure_ascii=True), encoding="utf-8"
        )

        # --- FAISS vector index (best-effort; skipped if optional deps missing) ---
        try:
            from ...agents.vector_rag import VectorRAGIndex  # noqa: PLC0415

            idx = VectorRAGIndex.load(
                namespace=resolved_namespace, rag_dir=DEFAULT_RAG_DIR
            )
            for chunk in payload["chunks"]:
                idx.ingest_text(
                    chunk["text"], source=source_identifier, tags=[resolved_namespace]
                )
            idx.save()
            logger.info(
                "[ui.ingest_reference_source] FAISS index updated namespace={} chunks={}",
                resolved_namespace,
                len(chunks),
            )
        except ImportError:
            logger.debug(
                "[ui.ingest_reference_source] FAISS not available; skipping vector index"
            )
        except Exception as faiss_exc:
            logger.warning(
                "[ui.ingest_reference_source] FAISS indexing failed (non-fatal): {}",
                faiss_exc,
            )

        insert_evidence_link(
            session_id=session_id,
            source_type="rag_ingest",
            source_id=source_identifier,
            relevance_score=0.88,
            rationale=f"Ingested {len(chunks)} chunk(s) into namespace '{resolved_namespace}'.",
            payload_json=json.dumps(
                {
                    "namespace": resolved_namespace,
                    "index_path": str(output_path.resolve()),
                    "chunk_count": len(chunks),
                },
                ensure_ascii=True,
            ),
            db_path=db_path,
        )
        merge_metadata(
            session_id,
            db_path=db_path,
            rag_source_path=source_identifier,
            rag_namespace=resolved_namespace,
            rag_status=f"Ingested {len(chunks)} chunk(s) from {source_label}.",
            rag_index_path=str(output_path.resolve()),
            rag_chunk_count=len(chunks),
            rag_provenance_text=(
                f"Namespace {resolved_namespace} | source {source_label} | chunks {len(chunks)}"
            ),
            latest_message=f"Reference source {source_label} ingested for retrieval.",
            latest_error_text="",
            remediation_hint="",
        )
        insert_tool_call_record(
            session_id=session_id,
            tool_name="ui.ingest_reference_source",
            input_json=json.dumps(
                {
                    "source_path": source_identifier,
                    "namespace": resolved_namespace,
                    "chunk_size": chunk_size,
                    "overlap": overlap,
                },
                ensure_ascii=True,
            ),
            output_json=json.dumps(
                {"index_path": str(output_path.resolve()), "chunk_count": len(chunks)},
                ensure_ascii=True,
            ),
            success=True,
            db_path=db_path,
        )
    except Exception as exc:
        merge_metadata(
            session_id,
            db_path=db_path,
            rag_source_path=source_reference,
            rag_namespace=resolved_namespace,
            rag_status="Reference ingestion failed.",
            latest_error_text=str(exc),
            remediation_hint=(
                "Use a readable local file or http/https URL and ensure optional "
                "PDF dependencies are installed."
            ),
        )
        insert_tool_call_record(
            session_id=session_id,
            tool_name="ui.ingest_reference_source",
            input_json=json.dumps(
                {"source_path": source_reference, "namespace": resolved_namespace},
                ensure_ascii=True,
            ),
            output_json=json.dumps({"error": str(exc)}, ensure_ascii=True),
            success=False,
            db_path=db_path,
        )

    return build_dashboard_state(session_id, db_path=db_path)

inspect_family async

inspect_family(session_id: str, user_goal: str, *, db_path: Path | None = None, model_name: str | None = None) -> dict[str, Any]

Run LLM-backed family classification and suggested checkpoints.

Parameters:

Name	Type	Description	Default
session_id	str	Dashboard session identifier.	required
user_goal	str	Free-text description of the design intent.	required
db_path	Path | None	Optional SQLite path override.	None
model_name	str | None	Optional provider-qualified model override.	None

Returns:

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

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

async def inspect_family(
    session_id: str,
    user_goal: str,
    *,
    db_path: Path | None = None,
    model_name: str | None = None,
) -> dict[str, Any]:
    """Run LLM-backed family classification and suggested checkpoints.

    Args:
        session_id: Dashboard session identifier.
        user_goal: Free-text description of the design intent.
        db_path: Optional SQLite path override.
        model_name: Optional provider-qualified model override.

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

    ensure_dashboard_session(session_id, user_goal=user_goal, db_path=db_path)
    session_row = get_design_session(session_id, db_path=db_path) or {}
    meta = _parse_json_blob(session_row.get("metadata_json"))
    resolved_model = normalize_model_name_for_provider(
        model_name or meta.get("model_name"),
        provider=sanitize_ui_text(meta.get("model_provider"), "github"),
        profile=sanitize_ui_text(meta.get("model_profile"), "balanced"),
    )
    resolved_endpoint = sanitize_ui_text(
        meta.get("local_endpoint"),
        os.getenv("SOLIDWORKS_UI_LOCAL_ENDPOINT", "http://127.0.0.1:11434/v1"),
    )

    local_family = meta.get("proposed_family", "") or "<not yet classified>"
    local_evidence = " | ".join(meta.get("family_evidence", [])) or "<none>"
    prompt = (
        "## TASK\n"
        "Apply the Feature-Tree-Reconstruction skill to classify the SolidWorks feature family "
        "and produce a human-reviewable checkpoint plan.\n\n"
        "## DESIGN GOAL\n"
        f"{user_goal}\n\n"
        "## MODEL CONTEXT\n"
        f"Active model path  : {meta.get('active_model_path', '') or '<none>'}\n"
        f"Active model status: {meta.get('active_model_status', '') or '<none>'}\n"
        f"Feature target refs: {meta.get('feature_target_text', '') or '<none>'}\n"
        f"Feature target status: {meta.get('feature_target_status', '') or '<none>'}\n\n"
        "## LOCAL CLASSIFIER EVIDENCE (pre-computed)\n"
        f"Family  : {local_family}\n"
        f"Evidence: {local_evidence}\n\n"
        "## REFERENCE CORPUS\n"
        f"{meta.get('rag_provenance_text', '') or '<none>'}\n\n"
        "## FEATURE-TREE RECONSTRUCTION SKILL\n"
        "Inspection sequence when model is available (use your mcp tools):\n"
        "  open_model → get_model_info → list_features(include_suppressed=True) "
        "→ get_mass_properties → classify_feature_tree\n"
        "Feature families: revolve | extrude | sheet_metal | advanced_solid | assembly | drawing | unknown\n"
        "Delegation rules:\n"
        "  - sheet_metal or advanced_solid → VBA-aware reconstruction path\n"
        "  - simple part family → direct MCP checkpoint plan\n"
        "  - assembly → component-first decomposition, part-level plan per component\n"
        "Guardrail: never reconstruct from silhouette only. "
        "If confidence is low and contradictory evidence exists, propose more inspection steps. "
        "When no model is attached, treat explicit user-supplied dimensions, named sketch phases, feature ordering, and manufacturing constraints as valid user-confirmed evidence for a conservative plan.\n\n"
        "## OUTPUT CONTRACT\n"
        "Return: family, confidence (high/medium/low), evidence[], warnings[], checkpoints[3-6].\n"
        "You must always emit at least 3 checkpoints when the prompt contains enough explicit geometry to start from an empty part, even if confidence is only low or medium.\n"
        "Each checkpoint: title, allowed_tools[] (from MCP tool catalog), rationale."
    )

    result = await _run_structured_agent(
        system_prompt=(
            "## ROLE\n"
            "You are a SolidWorks routing assistant applying the Feature-Tree-Reconstruction skill.\n"
            "Classify the feature family with evidence and confidence, then produce a safe "
            "checkpoint plan for human review.\n\n"
            "## ORCHESTRATION NOTES\n"
            "  - Inspection before planning: never produce a build plan without at least one "
            "evidence item from model inspection or user-confirmed context.\n"
            "  - Explicit user-specified dimensions, feature names, and operation ordering count as user-confirmed context when no model is attached.\n"
            "  - Propose 3-6 conservative checkpoints. Require human confirmation before each "
            "irreversible step.\n"
            "  - For sheet metal or unsupported advanced features, route to VBA-aware planning.\n"
            "  - Surface warnings when evidence is contradictory or confidence is low.\n"
            "  - Prefer 'extrude' for prompt-only parts built from named sketches and base extrusions unless stronger contrary evidence exists."
        ),
        user_prompt=prompt,
        result_type=FamilyInspection,
        model_name=resolved_model,
        local_endpoint=resolved_endpoint,
    )

    if isinstance(result, RecoverableFailure):
        ordered_features = _extract_explicit_feature_order(user_goal)
        if ordered_features:
            result = _coerce_explicit_feature_order_plan(
                user_goal,
                FamilyInspection(
                    family="extrude",
                    confidence="medium",
                    evidence=[
                        "Deterministic fallback engaged because model routing failed.",
                        "Goal includes an explicit feature-order sequence.",
                    ],
                    warnings=[
                        result.explanation,
                        "Using explicit feature-order checkpoints until model routing is healthy.",
                    ],
                    checkpoints=[],
                ),
            )
        else:
            merge_metadata(
                session_id,
                db_path=db_path,
                latest_message=result.explanation,
                latest_error_text=result.explanation,
                remediation_hint=(
                    result.remediation_steps[0]
                    if result.remediation_steps
                    else "Adjust provider/model settings, then retry inspect."
                ),
            )
            insert_tool_call_record(
                session_id=session_id,
                tool_name="ui.inspect_family",
                input_json=json.dumps({"user_goal": user_goal}, ensure_ascii=True),
                output_json=result.model_dump_json(),
                success=False,
                db_path=db_path,
            )
            return build_dashboard_state(session_id, db_path=db_path)

    result = _coerce_explicit_feature_order_plan(user_goal, result)

    evidence_payload = []
    for index, line in enumerate(result.evidence, start=1):
        insert_evidence_link(
            session_id=session_id,
            source_type="llm",
            source_id=f"family_evidence_{index}",
            relevance_score=0.85,
            rationale=line,
            payload_json=json.dumps({"family": result.family}, ensure_ascii=True),
            db_path=db_path,
        )
        evidence_payload.append(line)

    if result.checkpoints:
        replacement_rows: list[dict[str, Any]] = []
        for index, checkpoint in enumerate(result.checkpoints, start=1):
            replacement_rows.append(
                {
                    "checkpoint_index": index,
                    "title": checkpoint.title,
                    "planned_action_json": json.dumps(
                        {
                            "title": checkpoint.title,
                            "goal": checkpoint.title,
                            "tools": checkpoint.allowed_tools,
                            "rationale": checkpoint.rationale,
                            **checkpoint.execution,
                        },
                        ensure_ascii=True,
                    ),
                    "approved_by_user": index == 1,
                }
            )

        replace_plan_checkpoints(
            session_id=session_id,
            checkpoints=replacement_rows,
            db_path=db_path,
        )

    metadata = merge_metadata(
        session_id,
        db_path=db_path,
        user_goal=user_goal,
        proposed_family=result.family,
        family_confidence=result.confidence,
        family_evidence=evidence_payload,
        family_warnings=result.warnings,
        latest_message=f"Updated family classification to '{result.family}' from GitHub Copilot.",
        latest_error_text="",
        remediation_hint="",
    )
    insert_tool_call_record(
        session_id=session_id,
        tool_name="ui.inspect_family",
        input_json=json.dumps({"user_goal": user_goal}, ensure_ascii=True),
        output_json=result.model_dump_json(),
        success=True,
        db_path=db_path,
    )
    insert_evidence_link(
        session_id=session_id,
        source_type="llm",
        source_id="family_inspection",
        relevance_score=0.93,
        rationale="LLM family classification and checkpoint suggestions from GitHub Copilot.",
        payload_json=json.dumps(metadata, ensure_ascii=True),
        db_path=db_path,
    )
    return build_dashboard_state(session_id, db_path=db_path)

is_url_reference

is_url_reference(source_path: str) -> bool

Return True when source_path is an http/https URL.

Parameters:

Name	Type	Description	Default
source_path	str	Raw source path string from the UI.	required

Returns:

Type	Description
bool	True if the path starts with http:// or https://.

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

def is_url_reference(source_path: str) -> bool:
    """Return ``True`` when *source_path* is an http/https URL.

    Args:
        source_path: Raw source path string from the UI.

    Returns:
        ``True`` if the path starts with http:// or https://.
    """
    parsed = urlparse((source_path or "").strip())
    return parsed.scheme in {"http", "https"} and bool(parsed.netloc)

load_session_context

load_session_context(session_id: str, *, context_file: str | None = None, db_path: Path | None = None, context_dir: Path | None = None, api_origin: str = DEFAULT_API_ORIGIN) -> dict[str, Any]

Load a previously saved context snapshot back into session metadata.

Selectively merges the workflow_mode, assumptions_text, model path, feature targets, provider settings, notes, and docs query from the snapshot. Does not overwrite live execution state or checkpoint results.

Parameters:

Name	Type	Description	Default
session_id	str	Dashboard session identifier.	required
context_file	str | None	Path to the JSON snapshot file; defaults to the canonical location.	None
db_path	Path | None	Optional override for the SQLite database path.	None
context_dir	Path | None	Override directory for context snapshot files.	None
api_origin	str	API origin used for URL generation.	DEFAULT_API_ORIGIN

Returns:

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

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

def load_session_context(
    session_id: str,
    *,
    context_file: str | None = None,
    db_path: Path | None = None,
    context_dir: Path | None = None,
    api_origin: str = DEFAULT_API_ORIGIN,
) -> dict[str, Any]:
    """Load a previously saved context snapshot back into session metadata.

    Selectively merges the ``workflow_mode``, ``assumptions_text``, model path,
    feature targets, provider settings, notes, and docs query from the snapshot.
    Does not overwrite live execution state or checkpoint results.

    Args:
        session_id: Dashboard session identifier.
        context_file: Path to the JSON snapshot file; defaults to the canonical location.
        db_path: Optional override for the SQLite database path.
        context_dir: Override directory for context snapshot files.
        api_origin: API origin used for URL generation.

    Returns:
        Full dashboard state payload.
    """
    context_file_text = sanitize_ui_text(context_file, "")
    source_path = (
        Path(context_file_text)
        if context_file_text
        else context_file_path(session_id, context_dir=context_dir)
    )
    if not source_path.exists():
        message = f"Context load failed. File not found: {source_path}."
        merge_metadata(
            session_id,
            db_path=db_path,
            context_load_status=message,
            context_file_input=str(source_path),
            latest_error_text=message,
            remediation_hint="Save context first or provide a valid context file path.",
        )
        return build_dashboard_state(session_id, db_path=db_path, api_origin=api_origin)

    try:
        snapshot_payload = json.loads(source_path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        message = f"Context load failed: {exc}."
        merge_metadata(
            session_id,
            db_path=db_path,
            context_load_status=message,
            context_file_input=str(source_path),
            latest_error_text=message,
            remediation_hint="Ensure the context file is valid JSON saved by this dashboard.",
        )
        return build_dashboard_state(session_id, db_path=db_path, api_origin=api_origin)

    loaded_state = (
        snapshot_payload.get("state") if isinstance(snapshot_payload, dict) else {}
    )
    if not isinstance(loaded_state, dict):
        loaded_state = {}

    session_row = ensure_dashboard_session(session_id, db_path=db_path)
    metadata = parse_json_blob(session_row.get("metadata_json"))
    _RESTORABLE_KEYS = [
        "workflow_mode",
        "assumptions_text",
        "active_model_path",
        "active_model_status",
        "feature_target_text",
        "feature_target_status",
        "normalized_brief",
        "user_clarification_answer",
        "model_provider",
        "model_profile",
        "model_name",
        "local_endpoint",
        "rag_source_path",
        "rag_namespace",
        "notes_text",
        "docs_query",
        "docs_context_text",
    ]
    for key in _RESTORABLE_KEYS:
        if key in loaded_state:
            metadata[key] = loaded_state.get(key)

    upsert_design_session(
        session_id=session_id,
        user_goal=sanitize_ui_text(
            loaded_state.get("user_goal"),
            session_row.get("user_goal") or DEFAULT_USER_GOAL,
        ),
        source_mode=session_row.get("source_mode") or DEFAULT_SOURCE_MODE,
        accepted_family=(
            sanitize_ui_text(loaded_state.get("accepted_family"), "")
            or session_row.get("accepted_family")
        ),
        status=session_row.get("status") or "active",
        current_checkpoint_index=session_row.get("current_checkpoint_index") or 0,
        metadata_json=json.dumps(metadata, ensure_ascii=True),
        db_path=db_path,
    )

    message = f"Context loaded from {source_path}."
    persist_ui_action(
        session_id,
        tool_name="ui.context.load",
        db_path=db_path,
        metadata_updates={
            "context_load_status": message,
            "context_name_input": safe_context_name(source_path.stem, session_id),
            "context_file_input": str(source_path),
            "last_context_file": str(source_path),
            "latest_message": message,
            "latest_error_text": "",
            "remediation_hint": "",
        },
        input_payload={"context_file": str(source_path)},
    )
    return build_dashboard_state(session_id, db_path=db_path, api_origin=api_origin)

materialize_uploaded_model

materialize_uploaded_model(session_id: str, uploaded_files: list[dict[str, Any]] | None, *, upload_dir: Path | None = None) -> Path

Decode a base64-encoded uploaded model file and write it to the staging directory.

Parameters:

Name	Type	Description	Default
session_id	str	Dashboard session identifier used as a staging subdirectory.	required
uploaded_files	list[dict[str, Any]] | None	List of file payload dicts, each with name and data fields.	required
upload_dir	Path | None	Override for the upload staging directory.	None

Returns:

Type	Description
Path	Path pointing at the decoded file on disk.

Raises:

Type	Description
RuntimeError	When no file is provided, the name is missing, the suffix is unsupported, or the data field is not valid base64.

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

def materialize_uploaded_model(
    session_id: str,
    uploaded_files: list[dict[str, Any]] | None,
    *,
    upload_dir: Path | None = None,
) -> Path:
    """Decode a base64-encoded uploaded model file and write it to the staging directory.

    Args:
        session_id: Dashboard session identifier used as a staging subdirectory.
        uploaded_files: List of file payload dicts, each with ``name`` and ``data`` fields.
        upload_dir: Override for the upload staging directory.

    Returns:
        Path pointing at the decoded file on disk.

    Raises:
        RuntimeError: When no file is provided, the name is missing, the suffix is
            unsupported, or the data field is not valid base64.
    """
    if not uploaded_files:
        raise RuntimeError("No uploaded model file was provided.")

    upload = uploaded_files[0]
    file_name = Path(str(upload.get("name") or "")).name
    if not file_name:
        raise RuntimeError("Uploaded model is missing a filename.")

    suffix = Path(file_name).suffix.lower()
    if suffix not in SUPPORTED_MODEL_UPLOAD_SUFFIXES:
        allowed = ", ".join(sorted(SUPPORTED_MODEL_UPLOAD_SUFFIXES))
        raise RuntimeError(
            f"Unsupported uploaded model type: {suffix or 'unknown'}. Expected one of {allowed}."
        )

    encoded_data = upload.get("data")
    if not isinstance(encoded_data, str) or not encoded_data.strip():
        raise RuntimeError("Uploaded model is missing file data.")

    try:
        file_bytes = base64.b64decode(encoded_data, validate=True)
    except (binascii.Error, ValueError) as exc:
        raise RuntimeError("Uploaded model payload is not valid base64 data.") from exc

    target_dir = ensure_uploaded_model_dir(upload_dir) / session_id
    target_dir.mkdir(parents=True, exist_ok=True)
    target_path = target_dir / file_name
    target_path.write_bytes(file_bytes)
    return target_path

merge_metadata

merge_metadata(session_id: str, *, db_path: Path | None = None, user_goal: str | None = None, **updates: Any) -> dict[str, Any]

Read session metadata, merge updates into it, and write it back.

Implements the optimistic read-modify-write pattern used across all service functions that need to update one or more metadata keys without overwriting unrelated keys.

Parameters:

Name	Type	Description	Default
session_id	str	Target session identifier.	required
db_path	Path | None	Optional override for the SQLite database path.	None
user_goal	str | None	When provided, also updates the user_goal column.	None
**updates	Any	Arbitrary key-value pairs to merge into metadata.	{}

Returns:

Type	Description
dict[str, Any]	The merged metadata dict after the write.

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

def merge_metadata(
    session_id: str,
    *,
    db_path: Path | None = None,
    user_goal: str | None = None,
    **updates: Any,
) -> dict[str, Any]:
    """Read session metadata, merge *updates* into it, and write it back.

    Implements the optimistic read-modify-write pattern used across all
    service functions that need to update one or more metadata keys without
    overwriting unrelated keys.

    Args:
        session_id: Target session identifier.
        db_path: Optional override for the SQLite database path.
        user_goal: When provided, also updates the ``user_goal`` column.
        **updates: Arbitrary key-value pairs to merge into metadata.

    Returns:
        The merged metadata dict after the write.
    """
    session_row = get_design_session(session_id, db_path=db_path)
    metadata = parse_json_blob(session_row["metadata_json"]) if session_row else {}
    metadata.update(updates)

    effective_goal = user_goal or (
        session_row["user_goal"] if session_row else DEFAULT_USER_GOAL
    )
    effective_source = (
        session_row["source_mode"] if session_row else DEFAULT_SOURCE_MODE
    )
    effective_family = session_row["accepted_family"] if session_row else None
    effective_status = session_row["status"] if session_row else "active"
    effective_index = session_row["current_checkpoint_index"] if session_row else 0

    upsert_design_session(
        session_id=session_id,
        user_goal=effective_goal,
        source_mode=effective_source,
        accepted_family=effective_family,
        status=effective_status,
        current_checkpoint_index=effective_index,
        metadata_json=json.dumps(metadata, ensure_ascii=True),
        db_path=db_path,
    )
    return metadata

normalize_feature_targets

normalize_feature_targets(feature_target_text: str | None) -> list[str]

Parse a comma- or newline-separated feature target string into a list.

Strips @ prefixes and filters out tokens that look like file paths.

Parameters:

Name	Type	Description	Default
feature_target_text	str | None	Raw input such as "@Boss-Extrude1, @Sketch2".	required

Returns:

Type	Description
list[str]	List of normalised feature name strings.

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

def normalize_feature_targets(feature_target_text: str | None) -> list[str]:
    """Parse a comma- or newline-separated feature target string into a list.

    Strips ``@`` prefixes and filters out tokens that look like file paths.

    Args:
        feature_target_text: Raw input such as ``"@Boss-Extrude1, @Sketch2"``.

    Returns:
        List of normalised feature name strings.
    """
    targets: list[str] = []
    for raw in (feature_target_text or "").replace("\n", ",").split(","):
        normalized = raw.strip()
        if not normalized:
            continue
        candidate = normalized[1:] if normalized.startswith("@") else normalized
        if _looks_like_path_token(candidate):
            continue
        targets.append(candidate)
    return targets

normalize_workflow_mode

normalize_workflow_mode(workflow_mode: str | None) -> str

Normalise a raw workflow mode string to a known value.

Parameters:

Name	Type	Description	Default
workflow_mode	str | None	Raw mode string from UI state.	required

Returns:

Type	Description
str	"edit_existing", "new_design", or DEFAULT_WORKFLOW_MODE.

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

def normalize_workflow_mode(workflow_mode: str | None) -> str:
    """Normalise a raw workflow mode string to a known value.

    Args:
        workflow_mode: Raw mode string from UI state.

    Returns:
        ``"edit_existing"``, ``"new_design"``, or ``DEFAULT_WORKFLOW_MODE``.
    """
    normalized = (workflow_mode or DEFAULT_WORKFLOW_MODE).strip().lower()
    if normalized in {"edit_existing", "new_design"}:
        return normalized
    return DEFAULT_WORKFLOW_MODE

open_target_model async

open_target_model(session_id: str, *, model_path: str | None = None, uploaded_files: list[dict[str, Any]] | None = None, feature_target_text: str | None = None, db_path: Path | None = None, api_origin: str = DEFAULT_API_ORIGIN) -> dict[str, Any]

Open a target model in SolidWorks and persist session state.

Parameters:

Name	Type	Description	Default
session_id	str	Dashboard session identifier.	required
model_path	str | None	Absolute path to the SolidWorks model file.	None
uploaded_files	list[dict[str, Any]] | None	List of base64-encoded file upload dicts.	None
feature_target_text	str | None	Optional comma-separated feature target references.	None
db_path	Path | None	Optional SQLite path override.	None
api_origin	str	Base URL of the running FastAPI server.	DEFAULT_API_ORIGIN

Returns:

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

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

async def open_target_model(
    session_id: str,
    *,
    model_path: str | None = None,
    uploaded_files: list[dict[str, Any]] | None = None,
    feature_target_text: str | None = None,
    db_path: Path | None = None,
    api_origin: str = DEFAULT_API_ORIGIN,
) -> dict[str, Any]:
    """Open a target model in SolidWorks and persist session state.

    Args:
        session_id: Dashboard session identifier.
        model_path: Absolute path to the SolidWorks model file.
        uploaded_files: List of base64-encoded file upload dicts.
        feature_target_text: Optional comma-separated feature target references.
        db_path: Optional SQLite path override.
        api_origin: Base URL of the running FastAPI server.

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

    ensure_dashboard_session(session_id, db_path=db_path)
    adapter = None
    resolved_path: Path | None = None

    logger.info(
        "[ui.open_target_model] session_id={} model_path={} uploaded_files={} feature_targets={}",
        session_id,
        model_path,
        len(uploaded_files) if uploaded_files else 0,
        feature_target_text or "",
    )

    resolved_path = _resolve_model_path(
        session_id,
        model_path=model_path,
        uploaded_files=uploaded_files,
        feature_target_text=feature_target_text,
        db_path=db_path,
        api_origin=api_origin,
    )
    if resolved_path is None:
        return build_dashboard_state(session_id, db_path=db_path, api_origin=api_origin)

    config = load_config()
    adapter = await create_adapter(config)
    model_info: dict[str, Any] = {}
    tool_input = {
        "model_path": str(resolved_path.resolve()),
        "uploaded_file_name": uploaded_files[0].get("name") if uploaded_files else None,
        "feature_target_text": feature_target_text or "",
    }
    try:
        await adapter.connect()
        open_result = await adapter.open_model(str(resolved_path.resolve()))
        if not open_result.is_success:
            raise RuntimeError(open_result.error or "Failed to open target model.")

        if hasattr(adapter, "get_model_info"):
            info_result = await adapter.get_model_info()
            if info_result.is_success and isinstance(info_result.data, dict):
                model_info = info_result.data

        metadata = merge_metadata(
            session_id,
            db_path=db_path,
            workflow_mode="edit_existing",
            active_model_path=str(resolved_path.resolve()),
            active_model_status=(
                f"Opened model: {resolved_path.name}"
                f" | type={model_info.get('type', 'unknown')}"
            ),
            active_model_type=str(model_info.get("type") or ""),
            active_model_configuration=str(
                model_info.get("configuration") or "Default"
            ),
            feature_target_text=feature_target_text or "",
            latest_message=f"Opened target model {resolved_path.name} in SolidWorks.",
            latest_error_text="",
            remediation_hint="",
        )
        insert_tool_call_record(
            session_id=session_id,
            tool_name="ui.open_target_model",
            input_json=json.dumps(tool_input, ensure_ascii=True),
            output_json=json.dumps(metadata, ensure_ascii=True),
            success=True,
            db_path=db_path,
        )
    except Exception as exc:
        logger.exception(
            "[ui.open_target_model] failed session_id={} path={} error={}",
            session_id,
            str(resolved_path.resolve()) if resolved_path else "<none>",
            exc,
        )
        merge_metadata(
            session_id,
            db_path=db_path,
            workflow_mode="edit_existing",
            active_model_path=str(resolved_path.resolve()),
            feature_target_text=feature_target_text or "",
            latest_message="Failed to open target model.",
            latest_error_text=str(exc),
            remediation_hint="Open SolidWorks, verify COM access, and retry with a valid .sldprt/.sldasm path.",
        )
        insert_tool_call_record(
            session_id=session_id,
            tool_name="ui.open_target_model",
            input_json=json.dumps(tool_input, ensure_ascii=True),
            output_json=json.dumps({"error": str(exc)}, ensure_ascii=True),
            success=False,
            db_path=db_path,
        )
    finally:
        if adapter is not None:
            try:
                await adapter.disconnect()
            except Exception:
                logger.debug("Adapter disconnect failed during open-model cleanup")

    return build_dashboard_state(session_id, db_path=db_path, api_origin=api_origin)

parse_json_blob

parse_json_blob(payload: str | None) -> dict[str, Any]

Parse a JSON string into a dict, returning an empty dict on any failure.

Parameters:

Name	Type	Description	Default
payload	str | None	Raw JSON string, or None.	required

Returns:

Type	Description
dict[str, Any]	Parsed dict, or {} if parsing fails.

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

def parse_json_blob(payload: str | None) -> dict[str, Any]:
    """Parse a JSON string into a dict, returning an empty dict on any failure.

    Args:
        payload: Raw JSON string, or ``None``.

    Returns:
        Parsed dict, or ``{}`` if parsing fails.
    """
    if not payload:
        return {}
    try:
        parsed = json.loads(payload)
    except json.JSONDecodeError:
        return {}
    return parsed if isinstance(parsed, dict) else {}

persist_ui_action

persist_ui_action(session_id: str, *, tool_name: str, db_path: Path | None = None, metadata_updates: dict[str, Any] | None = None, user_goal: str | None = None, input_payload: dict[str, Any] | None = None, output_payload: dict[str, Any] | None = None, output_metadata: bool = False, success: bool = True, checkpoint_id: int | None = None) -> dict[str, Any]

Persist metadata updates and a matching tool-call audit record atomically.

Combines :func:merge_metadata and insert_tool_call_record so callers can update session state and write an audit entry in a single call.

Parameters:

Name	Type	Description	Default
session_id	str	Target session identifier.	required
tool_name	str	Logical name for the audit record (e.g. "ui.approve_brief").	required
db_path	Path | None	Optional override for the SQLite database path.	None
metadata_updates	dict[str, Any] | None	Key-value pairs to merge into session metadata.	None
user_goal	str | None	When provided, also updates the user_goal column.	None
input_payload	dict[str, Any] | None	Dict serialised as the input_json audit column.	None
output_payload	dict[str, Any] | None	Dict serialised as the output_json audit column.	None
output_metadata	bool	When True, write the merged metadata as output_json.	False
success	bool	Whether the action succeeded.	True
checkpoint_id	int | None	Optional FK to the associated plan checkpoint.	None

Returns:

Type	Description
dict[str, Any]	The merged metadata dict after the write.

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

def persist_ui_action(
    session_id: str,
    *,
    tool_name: str,
    db_path: Path | None = None,
    metadata_updates: dict[str, Any] | None = None,
    user_goal: str | None = None,
    input_payload: dict[str, Any] | None = None,
    output_payload: dict[str, Any] | None = None,
    output_metadata: bool = False,
    success: bool = True,
    checkpoint_id: int | None = None,
) -> dict[str, Any]:
    """Persist metadata updates and a matching tool-call audit record atomically.

    Combines :func:`merge_metadata` and ``insert_tool_call_record`` so callers
    can update session state and write an audit entry in a single call.

    Args:
        session_id: Target session identifier.
        tool_name: Logical name for the audit record (e.g. ``"ui.approve_brief"``).
        db_path: Optional override for the SQLite database path.
        metadata_updates: Key-value pairs to merge into session metadata.
        user_goal: When provided, also updates the ``user_goal`` column.
        input_payload: Dict serialised as the ``input_json`` audit column.
        output_payload: Dict serialised as the ``output_json`` audit column.
        output_metadata: When ``True``, write the merged metadata as ``output_json``.
        success: Whether the action succeeded.
        checkpoint_id: Optional FK to the associated plan checkpoint.

    Returns:
        The merged metadata dict after the write.
    """
    merged_metadata: dict[str, Any] = {}
    if metadata_updates is not None or user_goal is not None:
        merged_metadata = merge_metadata(
            session_id,
            db_path=db_path,
            user_goal=user_goal,
            **(metadata_updates or {}),
        )

    record_output = merged_metadata if output_metadata else output_payload

    insert_tool_call_record(
        session_id=session_id,
        checkpoint_id=checkpoint_id,
        tool_name=tool_name,
        input_json=(
            json.dumps(input_payload, ensure_ascii=True)
            if input_payload is not None
            else None
        ),
        output_json=(
            json.dumps(record_output, ensure_ascii=True)
            if record_output is not None
            else None
        ),
        success=success,
        db_path=db_path,
    )
    return merged_metadata

provider_from_model_name

provider_from_model_name(model_name: str) -> str

Infer the provider prefix from a provider-qualified model name.

Parameters:

Name	Type	Description	Default
model_name	str	Model name such as "github:openai/gpt-4.1".	required

Returns:

Type	Description
str	Provider string: "github", "openai", "anthropic",
str	"local", or "custom".

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

def provider_from_model_name(model_name: str) -> str:
    """Infer the provider prefix from a provider-qualified model name.

    Args:
        model_name: Model name such as ``"github:openai/gpt-4.1"``.

    Returns:
        Provider string: ``"github"``, ``"openai"``, ``"anthropic"``,
        ``"local"``, or ``"custom"``.
    """
    if model_name.startswith("github:"):
        return "github"
    if model_name.startswith("openai:"):
        return "openai"
    if model_name.startswith("anthropic:"):
        return "anthropic"
    if model_name.startswith("local:"):
        return "local"
    return "custom"

provider_has_credentials

provider_has_credentials(model_name: str, local_endpoint: str | None = None) -> bool

Check whether the required credentials exist for the given model.

Parameters:

Name	Type	Description	Default
model_name	str	Provider-qualified model name.	required
local_endpoint	str | None	Local API endpoint URL (required for "local:" models).	None

Returns:

Type	Description
bool	True if the relevant API key / endpoint is configured.

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

def provider_has_credentials(
    model_name: str, local_endpoint: str | None = None
) -> bool:
    """Check whether the required credentials exist for the given model.

    Args:
        model_name: Provider-qualified model name.
        local_endpoint: Local API endpoint URL (required for ``"local:"`` models).

    Returns:
        ``True`` if the relevant API key / endpoint is configured.
    """
    provider = provider_from_model_name(model_name)
    if provider == "github":
        token = os.getenv("GITHUB_API_KEY") or os.getenv("GH_TOKEN")
        return bool(token)
    if provider == "openai":
        return bool(os.getenv("OPENAI_API_KEY"))
    if provider == "anthropic":
        return bool(os.getenv("ANTHROPIC_API_KEY"))
    if provider == "local":
        return bool(local_endpoint)
    return True

read_reference_source

read_reference_source(source_path: Path) -> str

Read the text content of a local file (PDF, markdown, text).

PDF extraction requires the optional pypdf package.

Parameters:

Name	Type	Description	Default
source_path	Path	Local file path to read.	required

Returns:

Type	Description
str	Extracted plain text content.

Raises:

Type	Description
RuntimeError	When a PDF is supplied but pypdf is not installed.

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

def read_reference_source(source_path: Path) -> str:
    """Read the text content of a local file (PDF, markdown, text).

    PDF extraction requires the optional ``pypdf`` package.

    Args:
        source_path: Local file path to read.

    Returns:
        Extracted plain text content.

    Raises:
        RuntimeError: When a PDF is supplied but ``pypdf`` is not installed.
    """
    try:
        PdfReader = import_module("pypdf").PdfReader
    except ImportError:
        PdfReader = None

    suffix = source_path.suffix.lower()
    if suffix == ".pdf":
        if PdfReader is None:
            raise RuntimeError(
                "Install pypdf to ingest PDF sources, or provide a text/markdown file instead."
            )
        reader = PdfReader(str(source_path))
        return "\n\n".join((page.extract_text() or "") for page in reader.pages).strip()

    return source_path.read_text(encoding="utf-8")

read_reference_url

read_reference_url(source_url: str) -> tuple[str, str]

Fetch content from a URL and return (text, label).

Handles HTML (stripped), PDF (requires pypdf), and plain text.

Parameters:

Name	Type	Description	Default
source_url	str	http/https URL to fetch.	required

Returns:

Type	Description
tuple[str, str]	Tuple of (plain_text_content, label) where label is derived from the URL path.

Raises:

Type	Description
RuntimeError	When a PDF is served but pypdf is not installed.

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

def read_reference_url(source_url: str) -> tuple[str, str]:
    """Fetch content from a URL and return (text, label).

    Handles HTML (stripped), PDF (requires ``pypdf``), and plain text.

    Args:
        source_url: http/https URL to fetch.

    Returns:
        Tuple of (plain_text_content, label) where label is derived from the URL path.

    Raises:
        RuntimeError: When a PDF is served but ``pypdf`` is not installed.
    """
    try:
        PdfReader = import_module("pypdf").PdfReader
    except ImportError:
        PdfReader = None

    request = Request(source_url, headers={"User-Agent": "SolidWorksMCP/1.0"})
    with urlopen(request, timeout=20) as response:
        content_type = response.headers.get_content_type()
        charset = response.headers.get_content_charset() or "utf-8"
        raw_bytes = response.read()

    parsed = urlparse(source_url)
    label = Path(parsed.path).name or parsed.netloc or source_url
    suffix = Path(parsed.path).suffix.lower()

    if content_type == "application/pdf" or suffix == ".pdf":
        if PdfReader is None:
            raise RuntimeError(
                "Install pypdf to ingest PDF sources, or provide a text, markdown, or HTML source instead."
            )
        reader = PdfReader(BytesIO(raw_bytes))
        text = "\n\n".join((page.extract_text() or "") for page in reader.pages).strip()
        return text, label

    decoded = raw_bytes.decode(charset, errors="ignore")
    if "html" in content_type or suffix in {".html", ".htm"}:
        parser = HTMLTextExtractor()
        parser.feed(decoded)
        return parser.text().strip(), label

    return decoded.strip(), label

reconcile_manual_edits

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

Compare the two most-recent snapshots and summarise any detected changes.

Parameters:

Name	Type	Description	Default
session_id	str	Dashboard session identifier.	required
db_path	Path | None	Optional override for the SQLite database path.	None

Returns:

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

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

def reconcile_manual_edits(
    session_id: str,
    *,
    db_path: Path | None = None,
) -> dict[str, Any]:
    """Compare the two most-recent snapshots and summarise any detected changes.

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

    Returns:
        Full dashboard state payload.
    """
    ensure_dashboard_session(session_id, db_path=db_path)
    snapshots = list_model_state_snapshots(session_id, db_path=db_path)

    if len(snapshots) < 2:
        message = (
            "Not enough snapshots yet. Capture another preview after manual edits."
        )
    else:
        latest = snapshots[0]
        previous = snapshots[1]
        changed = latest.get("state_fingerprint") != previous.get(
            "state_fingerprint"
        ) or latest.get("screenshot_path") != previous.get("screenshot_path")
        message = (
            "Detected manual changes. Options: accept edits, patch toward goal, or rollback."
            if changed
            else "No visual/state change detected since the last accepted snapshot."
        )

    persist_ui_action(
        session_id,
        tool_name="ui.reconcile_manual_edits",
        db_path=db_path,
        metadata_updates={"latest_message": message},
        output_payload={"message": message},
    )
    return build_dashboard_state(session_id, db_path=db_path)

refresh_preview async

refresh_preview(session_id: str, *, orientation: str = DEFAULT_PREVIEW_ORIENTATION, db_path: Path | None = None, preview_dir: Path | None = None, api_origin: str = DEFAULT_API_ORIGIN, adapter_override: Any | None = None, active_model_path_override: str | None = None, reopen_active_model: bool = True) -> dict[str, Any]

Export the current SolidWorks viewport to a PNG preview and GLB/STL for the 3D viewer.

Parameters:

Name	Type	Description	Default
session_id	str	Dashboard session identifier.	required
orientation	str	View orientation for the PNG export ("front", "top", "right", "isometric", "current").	DEFAULT_PREVIEW_ORIENTATION
db_path	Path | None	Optional SQLite path override.	None
preview_dir	Path | None	Override for the preview output directory.	None
api_origin	str	Base URL of the running FastAPI server.	DEFAULT_API_ORIGIN
adapter_override	Any | None	Pre-connected adapter (avoids creating a second connection when called from connect_target_model).	None
active_model_path_override	str | None	Override for the model path to reopen.	None
reopen_active_model	bool	When True, reopen the persisted active model path before exporting (default).	True

Returns:

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

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

async def refresh_preview(
    session_id: str,
    *,
    orientation: str = DEFAULT_PREVIEW_ORIENTATION,
    db_path: Path | None = None,
    preview_dir: Path | None = None,
    api_origin: str = DEFAULT_API_ORIGIN,
    adapter_override: Any | None = None,
    active_model_path_override: str | None = None,
    reopen_active_model: bool = True,
) -> dict[str, Any]:
    """Export the current SolidWorks viewport to a PNG preview and GLB/STL for the 3D viewer.

    Args:
        session_id: Dashboard session identifier.
        orientation: View orientation for the PNG export
            (``"front"``, ``"top"``, ``"right"``, ``"isometric"``, ``"current"``).
        db_path: Optional SQLite path override.
        preview_dir: Override for the preview output directory.
        api_origin: Base URL of the running FastAPI server.
        adapter_override: Pre-connected adapter (avoids creating a second connection when
            called from ``connect_target_model``).
        active_model_path_override: Override for the model path to reopen.
        reopen_active_model: When ``True``, reopen the persisted active model path before
            exporting (default).

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

    ensure_dashboard_session(session_id, db_path=db_path)
    logger.info(
        "[ui.refresh_preview] session_id={} orientation={}",
        session_id,
        orientation,
    )
    session_row = get_design_session(session_id, db_path=db_path) or {}
    metadata = parse_json_blob(session_row.get("metadata_json"))
    preview_viewer_url = sanitize_preview_viewer_url(
        metadata.get("preview_viewer_url"),
        session_id=session_id,
        api_origin=api_origin,
    )
    resolved_preview_dir = ensure_preview_dir(preview_dir)
    preview_path = resolved_preview_dir / f"{session_id}.png"
    active_model_path = active_model_path_override or metadata.get("active_model_path")
    adapter = adapter_override
    owns_adapter = adapter_override is None

    try:
        if not active_model_path:
            raise RuntimeError(
                "No attached model path found for preview refresh. Attach a target model first."
            )

        if adapter is None:
            config = load_config()
            adapter = await create_adapter(config)
        if owns_adapter:
            await adapter.connect()
        if reopen_active_model and hasattr(adapter, "open_model"):
            await _reopen_target_model_for_preview(
                adapter, str(active_model_path), context="preview"
            )

        # --- Step 1: Export 3D geometry for the Three.js viewer (GLB preferred) ---
        glb_path = resolved_preview_dir / f"{session_id}.glb"
        stl_path = resolved_preview_dir / f"{session_id}.stl"
        viewer_ts = int(time.time())
        viewer_format = "none"
        try:
            glb_result = await adapter.export_file(str(glb_path.resolve()), "glb")
            if (
                glb_result.is_success
                and glb_path.exists()
                and glb_path.stat().st_size > 0
            ):
                viewer_format = "glb"
                viewer_ts = int(glb_path.stat().st_mtime)
                logger.info(
                    "[ui.refresh_preview] GLB export succeeded path={}",
                    str(glb_path.resolve()),
                )
        except Exception as _glb_exc:
            logger.warning("[ui.refresh_preview] GLB export failed: {}", str(_glb_exc))

        if viewer_format == "none":
            try:
                stl_result = await adapter.export_file(str(stl_path.resolve()), "stl")
                if (
                    stl_result.is_success
                    and stl_path.exists()
                    and stl_path.stat().st_size > 0
                ):
                    viewer_format = "stl"
                    viewer_ts = int(stl_path.stat().st_mtime)
                    logger.info(
                        "[ui.refresh_preview] STL export succeeded path={}",
                        str(stl_path.resolve()),
                    )
            except Exception:
                logger.debug("[ui.refresh_preview] STL export skipped (adapter error)")

        preview_viewer_url = (
            f"{api_origin}/api/ui/viewer/{session_id}?t={viewer_ts}&fmt={viewer_format}"
        )

        # --- Step 2: Export PNG screenshot (best-effort) ---
        png_payload = {
            "file_path": str(preview_path.resolve()),
            "format_type": "png",
            "width": 1280,
            "height": 720,
            "view_orientation": orientation,
        }
        png_ok = False
        png_error: str = ""
        snapshot_id: int | None = None
        try:
            result = await adapter.export_image(png_payload)
            if result.is_success and preview_path.exists():
                png_ok = True
                logger.info(
                    "[ui.refresh_preview] PNG export succeeded file_path={}",
                    str(preview_path.resolve()),
                )
                snapshot_id = insert_model_state_snapshot(
                    session_id=session_id,
                    screenshot_path=str(preview_path.resolve()),
                    state_fingerprint=f"preview-{preview_path.stat().st_mtime_ns}",
                    db_path=db_path,
                )
                insert_tool_call_record(
                    session_id=session_id,
                    tool_name="export_image",
                    input_json=json.dumps(png_payload, ensure_ascii=True),
                    output_json=json.dumps(result.data or {}, ensure_ascii=True),
                    success=True,
                    db_path=db_path,
                )
            else:
                png_error = result.error or "export_image returned failure"
                logger.warning("[ui.refresh_preview] PNG export failed: {}", png_error)
        except Exception as png_exc:
            png_error = str(png_exc)
            logger.warning("[ui.refresh_preview] PNG export exception: {}", png_exc)

        if owns_adapter:
            await adapter.disconnect()

        # --- Step 3: Export per-orientation PNG thumbnails ---
        VIEW_ORIENTATIONS = ["isometric", "front", "top", "right"]
        preview_view_urls: dict[str, str] = {}
        try:
            config2 = load_config()
            adapter2 = await create_adapter(config2)
            await adapter2.connect()
            if hasattr(adapter2, "open_model"):
                await _reopen_target_model_for_preview(
                    adapter2,
                    str(active_model_path),
                    context="orientation previews",
                )
            _sel_name = str(
                metadata.get("selected_feature_selector_name")
                or metadata.get("selected_feature_name")
                or ""
            ).strip()
            if _sel_name and hasattr(adapter2, "select_feature"):
                try:
                    await adapter2.select_feature(_sel_name)
                    logger.info(
                        "[ui.refresh_preview] re-selected '{}' before view screenshots",
                        _sel_name,
                    )
                except Exception as _sel_exc:
                    logger.debug(
                        "[ui.refresh_preview] re-select '{}' failed (non-fatal): {}",
                        _sel_name,
                        _sel_exc,
                    )
            for view_name in VIEW_ORIENTATIONS:
                view_path = resolved_preview_dir / f"{session_id}-{view_name}.png"
                try:
                    if _sel_name and hasattr(adapter2, "select_feature"):
                        await adapter2.select_feature(_sel_name)
                    view_result = await adapter2.export_image(
                        {
                            "file_path": str(view_path.resolve()),
                            "format_type": "png",
                            "width": 640,
                            "height": 480,
                            "view_orientation": view_name,
                        }
                    )
                    if view_result.is_success and view_path.exists():
                        ts = int(view_path.stat().st_mtime)
                        preview_view_urls[view_name] = (
                            f"{api_origin}/previews/{view_path.name}?ts={ts}"
                        )
                        logger.info(
                            "[ui.refresh_preview] view PNG {} exported", view_name
                        )
                    else:
                        logger.warning(
                            "[ui.refresh_preview] view PNG {} failed: {}",
                            view_name,
                            view_result.error or "no detail",
                        )
                except Exception as _ve:
                    logger.warning(
                        "[ui.refresh_preview] view PNG {} exception: {}",
                        view_name,
                        str(_ve),
                    )
            await adapter2.disconnect()
        except Exception as _views_exc:
            logger.warning(
                "[ui.refresh_preview] multi-view export failed: {}", str(_views_exc)
            )

        # Preserve existing view URLs when a refresh attempt returns no images.
        existing_view_urls = metadata.get("preview_view_urls")
        if isinstance(existing_view_urls, dict):
            if not preview_view_urls:
                preview_view_urls = dict(existing_view_urls)
            else:
                merged_view_urls = dict(existing_view_urls)
                merged_view_urls.update(preview_view_urls)
                preview_view_urls = merged_view_urls

        viewer_label = (
            f"3D viewer ({viewer_format.upper()})"
            if viewer_format != "none"
            else "3D viewer (no model)"
        )
        png_label = "PNG" if png_ok else f"no PNG ({png_error})"
        status_msg = f"Preview refreshed ({viewer_label}, {png_label})."

        merge_metadata(
            session_id,
            db_path=db_path,
            preview_orientation=orientation,
            latest_message=status_msg,
            preview_status=status_msg,
            latest_snapshot_id=(str(snapshot_id) if snapshot_id is not None else ""),
            preview_viewer_url=preview_viewer_url,
            preview_stl_ready=(viewer_format != "none"),
            preview_png_ready=png_ok,
            preview_view_urls=preview_view_urls,
            latest_error_text="",
            remediation_hint="",
        )
    except Exception as exc:
        logger.exception("[ui.refresh_preview] failed: {}", exc)
        insert_tool_call_record(
            session_id=session_id,
            tool_name="export_image",
            input_json=json.dumps({"orientation": orientation}, ensure_ascii=True),
            output_json=json.dumps({"error": str(exc)}, ensure_ascii=True),
            success=False,
            db_path=db_path,
        )
        merge_metadata(
            session_id,
            db_path=db_path,
            latest_message=f"Preview refresh failed: {exc}",
            preview_status=f"Preview refresh failed: {exc}",
            preview_orientation=orientation,
            latest_error_text=str(exc),
            remediation_hint="Open a model in SolidWorks and retry preview refresh.",
        )

    return build_dashboard_state(session_id, db_path=db_path, api_origin=api_origin)

request_clarifications async

request_clarifications(session_id: str, user_goal: str, *, user_answer: str = '', db_path: Path | None = None, model_name: str | None = None) -> dict[str, Any]

Generate focused follow-up questions for the current design goal using LLM.

Parameters:

Name	Type	Description	Default
session_id	str	Dashboard session identifier.	required
user_goal	str	Free-text description of the design intent.	required
user_answer	str	Any previous answers the user has already provided.	''
db_path	Path | None	Optional SQLite path override.	None
model_name	str | None	Optional provider-qualified model override.	None

Returns:

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

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

async def request_clarifications(
    session_id: str,
    user_goal: str,
    *,
    user_answer: str = "",
    db_path: Path | None = None,
    model_name: str | None = None,
) -> dict[str, Any]:
    """Generate focused follow-up questions for the current design goal using LLM.

    Args:
        session_id: Dashboard session identifier.
        user_goal: Free-text description of the design intent.
        user_answer: Any previous answers the user has already provided.
        db_path: Optional SQLite path override.
        model_name: Optional provider-qualified model override.

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

    ensure_dashboard_session(session_id, user_goal=user_goal, db_path=db_path)
    session_row = get_design_session(session_id, db_path=db_path) or {}
    meta = _parse_json_blob(session_row.get("metadata_json"))
    resolved_model = normalize_model_name_for_provider(
        model_name or meta.get("model_name"),
        provider=sanitize_ui_text(meta.get("model_provider"), "github"),
        profile=sanitize_ui_text(meta.get("model_profile"), "balanced"),
    )
    resolved_endpoint = sanitize_ui_text(
        meta.get("local_endpoint"),
        os.getenv("SOLIDWORKS_UI_LOCAL_ENDPOINT", "http://127.0.0.1:11434/v1"),
    )

    answer_section = (
        f"\n## USER ANSWERS / CLARIFICATIONS\n{user_answer}" if user_answer else ""
    )
    assumptions_text = sanitize_ui_text(
        meta.get("assumptions_text"), "<none specified>"
    )
    prompt = (
        "## TASK\n"
        "Prepare a SolidWorks design brief using the Printer-Profile-Tolerancing skill.\n"
        "Return a normalized_brief and at most three clarifying_questions that unblock the next modeling step.\n\n"
        "## DESIGN GOAL\n"
        f"{user_goal}\n\n"
        "## MANUFACTURING ASSUMPTIONS\n"
        f"{assumptions_text}\n\n"
        "## MODEL CONTEXT\n"
        f"Active model path : {meta.get('active_model_path', '') or '<none>'}\n"
        f"Feature target refs: {meta.get('feature_target_text', '') or '<none>'}\n"
        f"Reference corpus   : {meta.get('rag_provenance_text', '') or '<none>'}"
        f"{answer_section}\n\n"
        "## OUTPUT CONTRACT\n"
        "normalized_brief: concise paragraph with explicit dimensions/tolerances where known (≥10 chars).\n"
        "questions       : list of up to 3 highest-leverage questions that unblock modeling.\n"
        "  - Return an empty question list when the goal, assumptions, and user answers already provide enough detail to sketch and dimension the next feature.\n"
        "  - Include material/layer-height/nozzle values if missing from assumptions.\n"
        "  - Include critical fit/clearance targets if unspecified.\n"
        "  - Do not ask questions already answered above."
    )

    result = await _run_structured_agent(
        system_prompt=(
            "## ROLE\n"
            "You are a CAD planning assistant applying the Printer-Profile-Tolerancing skill.\n"
            "Normalize goals into manufacturing-ready language with explicit tolerance/clearance "
            "targets (e.g. '0.30 mm mating clearance', '0.2 mm layer height'). "
            "Ask only the highest-leverage questions that unblock the SolidWorks modeling steps. "
            "Always surface material, nozzle size, and orientation constraints when present in the goal. "
            "If the user has already supplied explicit dimensions, wall thickness, fit targets, and feature placement, do not keep restating the same asks; return zero questions instead."
        ),
        user_prompt=prompt,
        result_type=ClarificationResponse,
        model_name=resolved_model,
        local_endpoint=resolved_endpoint,
    )

    if isinstance(result, RecoverableFailure):
        merge_metadata(
            session_id,
            db_path=db_path,
            latest_message=result.explanation,
            clarifying_questions=[],
            latest_error_text=result.explanation,
            remediation_hint=(
                result.remediation_steps[0]
                if result.remediation_steps
                else "Configure provider credentials and retry."
            ),
        )
        insert_tool_call_record(
            session_id=session_id,
            tool_name="ui.request_clarifications",
            input_json=json.dumps({"user_goal": user_goal}, ensure_ascii=True),
            output_json=result.model_dump_json(),
            success=False,
            db_path=db_path,
        )
        return build_dashboard_state(session_id, db_path=db_path)

    metadata = merge_metadata(
        session_id,
        db_path=db_path,
        user_goal=user_goal,
        normalized_brief=result.normalized_brief,
        clarifying_questions=result.questions,
        user_clarification_answer=user_answer,
        latest_message="Generated clarifying questions from GitHub Copilot.",
        latest_error_text="",
        remediation_hint="",
    )
    insert_tool_call_record(
        session_id=session_id,
        tool_name="ui.request_clarifications",
        input_json=json.dumps({"user_goal": user_goal}, ensure_ascii=True),
        output_json=result.model_dump_json(),
        success=True,
        db_path=db_path,
    )
    insert_evidence_link(
        session_id=session_id,
        source_type="llm",
        source_id="clarification_response",
        relevance_score=0.9,
        rationale="Normalized brief and follow-up questions from GitHub Copilot.",
        payload_json=json.dumps(metadata, ensure_ascii=True),
        db_path=db_path,
    )
    return build_dashboard_state(session_id, db_path=db_path)

run_go_orchestration async

run_go_orchestration(session_id: str, *, user_goal: str, assumptions_text: str | None = None, user_answer: str = '', db_path: Path | None = None, api_origin: str = DEFAULT_API_ORIGIN) -> dict[str, Any]

Run a single end-to-end pass: approve brief, update preferences, clarify, inspect.

Parameters:

Name	Type	Description	Default
session_id	str	Dashboard session identifier.	required
user_goal	str	Free-text description of the design intent.	required
assumptions_text	str | None	Optional manufacturing assumptions to persist.	None
user_answer	str	Any previous answers the user has already provided.	''
db_path	Path | None	Optional SQLite path override.	None
api_origin	str	Base URL of the running FastAPI server.	DEFAULT_API_ORIGIN

Returns:

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

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

async def run_go_orchestration(
    session_id: str,
    *,
    user_goal: str,
    assumptions_text: str | None = None,
    user_answer: str = "",
    db_path: Path | None = None,
    api_origin: str = DEFAULT_API_ORIGIN,
) -> dict[str, Any]:
    """Run a single end-to-end pass: approve brief, update preferences, clarify, inspect.

    Args:
        session_id: Dashboard session identifier.
        user_goal: Free-text description of the design intent.
        assumptions_text: Optional manufacturing assumptions to persist.
        user_answer: Any previous answers the user has already provided.
        db_path: Optional SQLite path override.
        api_origin: Base URL of the running FastAPI server.

    Returns:
        Full dashboard state payload.
    """
    from .session_service import (  # noqa: PLC0415
        approve_design_brief,
        build_dashboard_state,
        update_ui_preferences,
    )

    try:
        goal_text = sanitize_ui_text(user_goal, DEFAULT_USER_GOAL)
        approve_design_brief(session_id, goal_text, db_path=db_path)

        session_row = get_design_session(session_id, db_path=db_path) or {}
        meta = _parse_json_blob(session_row.get("metadata_json"))
        update_ui_preferences(
            session_id,
            assumptions_text=assumptions_text,
            model_provider=str(meta.get("model_provider") or "github"),
            model_profile=str(meta.get("model_profile") or "balanced"),
            model_name=meta.get("model_name"),
            local_endpoint=meta.get("local_endpoint"),
            db_path=db_path,
        )

        await request_clarifications(
            session_id,
            goal_text,
            user_answer=user_answer,
            db_path=db_path,
        )
        await inspect_family(session_id, goal_text, db_path=db_path)

        persist_ui_action(
            session_id,
            tool_name="ui.orchestrate_go",
            db_path=db_path,
            metadata_updates={
                "orchestration_status": (
                    "Go run completed: inputs saved, clarifications refreshed, engineering review updated."
                ),
                "latest_message": "Go run completed across workflow, review, and model output lanes.",
                "latest_error_text": "",
                "remediation_hint": "",
            },
            input_payload={
                "user_goal": goal_text,
                "assumptions_text": assumptions_text,
                "user_answer": user_answer,
            },
            output_payload={
                "status": "success",
                "message": "Go orchestration completed.",
            },
        )
    except Exception as exc:
        logger.exception("[ui.run_go_orchestration] failed session_id={}", session_id)
        merge_metadata(
            session_id,
            db_path=db_path,
            orchestration_status="Go run failed.",
            latest_error_text=str(exc),
            remediation_hint="Review provider credentials/model selection and retry Go.",
        )
    return build_dashboard_state(session_id, db_path=db_path, api_origin=api_origin)

safe_context_name

safe_context_name(context_name: str | None, session_id: str) -> str

Normalise a context name to a safe filesystem slug.

Parameters:

Name	Type	Description	Default
context_name	str | None	User-supplied context name, may be None.	required
session_id	str	Fallback identifier.	required

Returns:

Type	Description
str	Alphanumeric + hyphen slug, never empty.

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

def safe_context_name(context_name: str | None, session_id: str) -> str:
    """Normalise a context name to a safe filesystem slug.

    Args:
        context_name: User-supplied context name, may be None.
        session_id: Fallback identifier.

    Returns:
        Alphanumeric + hyphen slug, never empty.
    """
    base = (context_name or session_id or "prefab-dashboard").strip()
    allowed = [ch if (ch.isalnum() or ch in {"-", "_"}) else "-" for ch in base]
    normalized = "".join(allowed).strip("-")
    return normalized or "prefab-dashboard"

sanitize_model_path_text

sanitize_model_path_text(value: Any) -> str

Strip surrounding quotes from a model path string.

Parameters:

Name	Type	Description	Default
value	Any	Raw model path value from UI state.	required

Returns:

Type	Description
str	Cleaned path string, or "" if empty.

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

def sanitize_model_path_text(value: Any) -> str:
    """Strip surrounding quotes from a model path string.

    Args:
        value: Raw model path value from UI state.

    Returns:
        Cleaned path string, or ``""`` if empty.
    """
    text = sanitize_ui_text(value, "")
    if len(text) >= 2 and text[0] == text[-1] and text[0] in {'"', "'"}:
        text = text[1:-1].strip()
    return text

sanitize_preview_viewer_url

sanitize_preview_viewer_url(value: Any, *, session_id: str, api_origin: str) -> str

Validate and return a preview viewer URL, or "" if it looks wrong.

Rejects URLs pointing at unexpected origins or session IDs to prevent open-redirect-style issues in the embedded viewer iframe.

Parameters:

Name	Type	Description	Default
value	Any	Raw URL string from session metadata.	required
session_id	str	Expected session ID segment in the path.	required
api_origin	str	Allowed API origin (scheme + host + port).	required

Returns:

Type	Description
str	Validated URL string, or "" if validation fails.

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

def sanitize_preview_viewer_url(
    value: Any,
    *,
    session_id: str,
    api_origin: str,
) -> str:
    """Validate and return a preview viewer URL, or ``""`` if it looks wrong.

    Rejects URLs pointing at unexpected origins or session IDs to prevent
    open-redirect-style issues in the embedded viewer iframe.

    Args:
        value: Raw URL string from session metadata.
        session_id: Expected session ID segment in the path.
        api_origin: Allowed API origin (scheme + host + port).

    Returns:
        Validated URL string, or ``""`` if validation fails.
    """
    text = sanitize_ui_text(value, "")
    if not text:
        return ""
    parsed = urlparse(text)
    path = (parsed.path or "").rstrip("/")
    expected_path = f"/api/ui/viewer/{session_id}".rstrip("/")
    if path != expected_path:
        return ""
    if parsed.scheme and parsed.netloc:
        expected = urlparse(api_origin)
        if parsed.netloc != expected.netloc:
            return ""
    return text

sanitize_ui_text

sanitize_ui_text(value: Any, fallback: str = '') -> str

Return a clean string from value, using fallback for empty/invalid inputs.

Strips template placeholders such as {{ field }}, bare quotes, and pydantic-ai expression strings that sometimes leak into UI state.

Parameters:

Name	Type	Description	Default
value	Any	Arbitrary input (str, None, etc.).	required
fallback	str	Value to return when value is empty or invalid.	''

Returns:

Type	Description
str	Cleaned string, or fallback.

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

def sanitize_ui_text(value: Any, fallback: str = "") -> str:
    """Return a clean string from *value*, using *fallback* for empty/invalid inputs.

    Strips template placeholders such as ``{{ field }}``, bare quotes, and
    pydantic-ai expression strings that sometimes leak into UI state.

    Args:
        value: Arbitrary input (str, None, etc.).
        fallback: Value to return when *value* is empty or invalid.

    Returns:
        Cleaned string, or *fallback*.
    """
    if value is None:
        return fallback
    text = str(value).strip()
    if not text:
        return fallback
    if text in {'"', "'"}:
        return fallback
    if text.startswith("{{") and text.endswith("}}"):
        return fallback
    if "$result." in text or "$error" in text:
        return fallback
    return text

save_session_context

save_session_context(session_id: str, *, context_name: str | None = None, db_path: Path | None = None, context_dir: Path | None = None, api_origin: str = DEFAULT_API_ORIGIN) -> dict[str, Any]

Persist the current dashboard state to a plain JSON snapshot file.

The snapshot can later be loaded with :func:load_session_context to restore the same session state across restarts or share it between machines.

Parameters:

Name	Type	Description	Default
session_id	str	Dashboard session identifier.	required
context_name	str | None	Optional human-readable name for the snapshot file.	None
db_path	Path | None	Optional override for the SQLite database path.	None
context_dir	Path | None	Override directory for context snapshot files.	None
api_origin	str	API origin used for URL generation.	DEFAULT_API_ORIGIN

Returns:

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

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

def save_session_context(
    session_id: str,
    *,
    context_name: str | None = None,
    db_path: Path | None = None,
    context_dir: Path | None = None,
    api_origin: str = DEFAULT_API_ORIGIN,
) -> dict[str, Any]:
    """Persist the current dashboard state to a plain JSON snapshot file.

    The snapshot can later be loaded with :func:`load_session_context` to restore
    the same session state across restarts or share it between machines.

    Args:
        session_id: Dashboard session identifier.
        context_name: Optional human-readable name for the snapshot file.
        db_path: Optional override for the SQLite database path.
        context_dir: Override directory for context snapshot files.
        api_origin: API origin used for URL generation.

    Returns:
        Full dashboard state payload.
    """
    state = build_dashboard_state(session_id, db_path=db_path, api_origin=api_origin)
    target_path = context_file_path(
        session_id,
        context_name=context_name,
        context_dir=context_dir,
    )
    payload = {
        "session_id": session_id,
        "saved_at": int(time.time()),
        "state": state,
    }
    target_path.write_text(
        json.dumps(payload, ensure_ascii=True, indent=2), encoding="utf-8"
    )
    message = f"Context saved to {target_path}."
    persist_ui_action(
        session_id,
        tool_name="ui.context.save",
        db_path=db_path,
        metadata_updates={
            "context_save_status": message,
            "context_name_input": safe_context_name(context_name, session_id),
            "context_file_input": str(target_path),
            "last_context_file": str(target_path),
            "latest_message": message,
            "latest_error_text": "",
            "remediation_hint": "",
        },
        input_payload={"context_name": context_name},
        output_payload={"path": str(target_path)},
    )
    return build_dashboard_state(session_id, db_path=db_path, api_origin=api_origin)

select_workflow_mode

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

Persist the onboarding workflow branch and reset new-design state when switching.

Selecting "new_design" resets all model-attachment, preview, and clarification state so the user starts with a blank slate for the new part.

Parameters:

Name	Type	Description	Default
session_id	str	Dashboard session identifier.	required
workflow_mode	str	Target mode ("edit_existing" or "new_design").	required
db_path	Path | None	Optional override for the SQLite database path.	None

Returns:

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

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

def select_workflow_mode(
    session_id: str,
    *,
    workflow_mode: str,
    db_path: Path | None = None,
) -> dict[str, Any]:
    """Persist the onboarding workflow branch and reset new-design state when switching.

    Selecting ``"new_design"`` resets all model-attachment, preview, and clarification
    state so the user starts with a blank slate for the new part.

    Args:
        session_id: Dashboard session identifier.
        workflow_mode: Target mode (``"edit_existing"`` or ``"new_design"``).
        db_path: Optional override for the SQLite database path.

    Returns:
        Full dashboard state payload.
    """
    session_row = ensure_dashboard_session(session_id, db_path=db_path)
    normalized_mode = normalize_workflow_mode(workflow_mode)
    workflow_label, workflow_guidance, _ = workflow_copy(normalized_mode)

    if normalized_mode == "new_design":
        metadata = parse_json_blob(session_row.get("metadata_json"))
        metadata.update(
            {
                "workflow_mode": normalized_mode,
                "active_model_path": "",
                "active_model_status": "No active model connected yet.",
                "active_model_type": "",
                "active_model_configuration": "",
                "feature_target_text": "",
                "feature_target_status": "No grounded feature target selected.",
                "selected_feature_name": "",
                "selected_feature_selector_name": "",
                "preview_viewer_url": "",
                "preview_view_urls": {},
                "preview_status": "No preview captured yet.",
                "preview_stl_ready": False,
                "preview_png_ready": False,
                "clarifying_questions": [],
                "user_clarification_answer": "",
                "proposed_family": "unclassified",
                "family_confidence": "pending",
                "family_evidence": [],
                "family_warnings": [],
                "mocked_tools": [],
                "rag_source_path": "",
                "rag_status": "No retrieval source ingested yet.",
                "rag_index_path": "",
                "rag_chunk_count": 0,
                "rag_provenance_text": "No retrieval provenance available yet.",
                "docs_context_text": "No docs context loaded yet.",
                "notes_text": "",
                "orchestration_status": "Ready.",
                "context_save_status": "",
                "context_load_status": "",
                "latest_message": f"Workflow selected: {workflow_label}.",
                "latest_error_text": "",
                "remediation_hint": "",
                "normalized_brief": "Describe the new part you want to design.",
            }
        )
        upsert_design_session(
            session_id=session_id,
            user_goal="Describe the new part you want to design.",
            source_mode=session_row.get("source_mode") or DEFAULT_SOURCE_MODE,
            accepted_family=None,
            status="inspect",
            current_checkpoint_index=0,
            metadata_json=json.dumps(metadata, ensure_ascii=True),
            db_path=db_path,
        )
        # Reset all checkpoints so Execute Next starts clean for the new design.
        for row in list_plan_checkpoints(session_id, db_path=db_path):
            update_plan_checkpoint(
                int(row.get("id") or 0),
                approved_by_user=False,
                executed=False,
                result_json="",
                db_path=db_path,
            )
    else:
        metadata = merge_metadata(
            session_id,
            db_path=db_path,
            workflow_mode=normalized_mode,
            latest_message=f"Workflow selected: {workflow_label}.",
            latest_error_text="",
            remediation_hint="",
        )

    persist_ui_action(
        session_id,
        tool_name="ui.select_workflow_mode",
        db_path=db_path,
        input_payload={"workflow_mode": normalized_mode},
        output_payload={
            "workflow_mode": normalized_mode,
            "workflow_label": workflow_label,
            "workflow_guidance_text": workflow_guidance,
            "metadata": metadata,
        },
    )
    return build_dashboard_state(session_id, db_path=db_path)

trace_json

trace_json(value: Any) -> str

Serialise value to a pretty-printed JSON string for operator trace panels.

Parameters:

Name	Type	Description	Default
value	Any	Any JSON-serialisable value.	required

Returns:

Type	Description
str	Indented JSON string.

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

def trace_json(value: Any) -> str:
    """Serialise *value* to a pretty-printed JSON string for operator trace panels.

    Args:
        value: Any JSON-serialisable value.

    Returns:
        Indented JSON string.
    """
    return json.dumps(value, ensure_ascii=True, indent=2, default=_trace_json_default)

trace_session_row

trace_session_row(session_row: dict[str, Any] | None) -> dict[str, Any]

Return a copy of session_row with the bulky metadata_json field removed.

Parameters:

Name	Type	Description	Default
session_row	dict[str, Any] | None	Raw session row dict from the database.	required

Returns:

Type	Description
dict[str, Any]	Filtered dict suitable for debug display.

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

def trace_session_row(session_row: dict[str, Any] | None) -> dict[str, Any]:
    """Return a copy of *session_row* with the bulky ``metadata_json`` field removed.

    Args:
        session_row: Raw session row dict from the database.

    Returns:
        Filtered dict suitable for debug display.
    """
    if not session_row:
        return {}
    return {key: value for key, value in session_row.items() if key != "metadata_json"}

trace_tool_records

trace_tool_records(records: list[dict[str, Any]]) -> list[dict[str, Any]]

Return the last ten tool-call records as lean trace dicts.

Parameters:

Name	Type	Description	Default
records	list[dict[str, Any]]	Full list of tool-call records for the session.	required

Returns:

Type	Description
list[dict[str, Any]]	Trimmed list of trace-friendly dicts (id, tool_name, success, …).

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

def trace_tool_records(records: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Return the last ten tool-call records as lean trace dicts.

    Args:
        records: Full list of tool-call records for the session.

    Returns:
        Trimmed list of trace-friendly dicts (id, tool_name, success, …).
    """
    traced: list[dict[str, Any]] = []
    for record in records[-10:]:
        traced.append(
            {
                "id": record.get("id"),
                "tool_name": record.get("tool_name"),
                "success": record.get("success"),
                "input_json": record.get("input_json"),
                "output_json": record.get("output_json"),
                "created_at": record.get("created_at"),
            }
        )
    return traced

update_session_notes

update_session_notes(session_id: str, *, notes_text: str, db_path: Path | None = None, api_origin: str = DEFAULT_API_ORIGIN) -> dict[str, Any]

Persist free-form engineering notes in session metadata.

Parameters:

Name	Type	Description	Default
session_id	str	Dashboard session identifier.	required
notes_text	str	Free-text engineering notes content.	required
db_path	Path | None	Optional override for the SQLite database path.	None
api_origin	str	API origin used for URL generation in the returned state.	DEFAULT_API_ORIGIN

Returns:

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

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

def update_session_notes(
    session_id: str,
    *,
    notes_text: str,
    db_path: Path | None = None,
    api_origin: str = DEFAULT_API_ORIGIN,
) -> dict[str, Any]:
    """Persist free-form engineering notes in session metadata.

    Args:
        session_id: Dashboard session identifier.
        notes_text: Free-text engineering notes content.
        db_path: Optional override for the SQLite database path.
        api_origin: API origin used for URL generation in the returned state.

    Returns:
        Full dashboard state payload.
    """
    persist_ui_action(
        session_id,
        tool_name="ui.notes.update",
        db_path=db_path,
        metadata_updates={
            "notes_text": notes_text,
            "latest_message": "Notes saved.",
            "latest_error_text": "",
            "remediation_hint": "",
        },
        input_payload={"notes_text": notes_text},
    )
    return build_dashboard_state(session_id, db_path=db_path, api_origin=api_origin)

update_ui_preferences

update_ui_preferences(session_id: str, *, assumptions_text: str | None = None, model_provider: str | None = None, model_profile: str | None = None, model_name: str | None = None, local_endpoint: str | None = None, db_path: Path | None = None) -> dict[str, Any]

Persist manufacturing assumptions and LLM provider preferences.

Parameters:

Name	Type	Description	Default
session_id	str	Dashboard session identifier.	required
assumptions_text	str | None	Free-form manufacturing assumption text.	None
model_provider	str | None	LLM provider identifier ("github", "openai", etc.).	None
model_profile	str | None	Capability tier ("small", "balanced", "large").	None
model_name	str | None	Explicit provider-qualified model name override.	None
local_endpoint	str | None	Ollama / local API base URL.	None
db_path	Path | None	Optional override for the SQLite database path.	None

Returns:

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

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

def update_ui_preferences(
    session_id: str,
    *,
    assumptions_text: str | None = None,
    model_provider: str | None = None,
    model_profile: str | None = None,
    model_name: str | None = None,
    local_endpoint: str | None = None,
    db_path: Path | None = None,
) -> dict[str, Any]:
    """Persist manufacturing assumptions and LLM provider preferences.

    Args:
        session_id: Dashboard session identifier.
        assumptions_text: Free-form manufacturing assumption text.
        model_provider: LLM provider identifier (``"github"``, ``"openai"``, etc.).
        model_profile: Capability tier (``"small"``, ``"balanced"``, ``"large"``).
        model_name: Explicit provider-qualified model name override.
        local_endpoint: Ollama / local API base URL.
        db_path: Optional override for the SQLite database path.

    Returns:
        Full dashboard state payload.
    """
    import os  # local import to keep module-level deps minimal

    ensure_dashboard_session(session_id, db_path=db_path)
    provider = (model_provider or "github").strip().lower()
    profile = (model_profile or "balanced").strip().lower()
    resolved_model = normalize_model_name_for_provider(
        model_name,
        provider=provider,
        profile=profile,
    )
    resolved_endpoint = sanitize_ui_text(
        local_endpoint,
        os.getenv("SOLIDWORKS_UI_LOCAL_ENDPOINT", "http://127.0.0.1:11434/v1"),
    )
    persist_ui_action(
        session_id,
        tool_name="ui.update_preferences",
        db_path=db_path,
        metadata_updates={
            "assumptions_text": sanitize_ui_text(
                assumptions_text,
                "No assumptions provided yet.",
            ),
            "model_provider": provider,
            "model_profile": profile,
            "model_name": resolved_model,
            "local_endpoint": resolved_endpoint,
            "latest_message": "Updated assumptions and model preferences.",
            "latest_error_text": "",
            "remediation_hint": "",
        },
        input_payload={
            "assumptions_text": assumptions_text,
            "model_provider": provider,
            "model_profile": profile,
            "model_name": resolved_model,
            "local_endpoint": resolved_endpoint,
        },
        output_metadata=True,
    )
    return build_dashboard_state(session_id, db_path=db_path)

workflow_copy

workflow_copy(workflow_mode: str, active_model_path: str | None = None) -> tuple[str, str, str]

Return display copy for the workflow selector.

Parameters:

Name	Type	Description	Default
workflow_mode	str	Normalised workflow mode string.	required
active_model_path	str | None	Currently attached model path, if any.	None

Returns:

Type	Description
tuple[str, str, str]	Three-tuple of (label, guidance_text, flow_header_text).

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

def workflow_copy(
    workflow_mode: str, active_model_path: str | None = None
) -> tuple[str, str, str]:
    """Return display copy for the workflow selector.

    Args:
        workflow_mode: Normalised workflow mode string.
        active_model_path: Currently attached model path, if any.

    Returns:
        Three-tuple of (label, guidance_text, flow_header_text).
    """
    has_active_model = bool(str(active_model_path or "").strip())
    if workflow_mode == "edit_existing":
        return (
            "Editing Existing Part or Assembly",
            "Attach an existing SolidWorks file, inspect the feature tree, then describe the feature edits you want.",
            "Choose Workflow -> Attach Model -> Inspect -> Plan -> Execute",
        )
    if workflow_mode == "new_design":
        return (
            "New Design From Scratch",
            "Define the design goal and assumptions first, then inspect the proposed family before executing CAD steps.",
            "Choose Workflow -> Define Goal -> Clarify -> Plan -> Execute",
        )
    if has_active_model:
        return (
            "Choose a Workflow",
            "An active model is already attached. Pick whether you want to keep editing that file or pivot to a new design flow.",
            "Choose Workflow -> Attach Model or Define Goal -> Inspect -> Plan -> Execute",
        )
    return (
        "Choose a Workflow",
        "Choose whether you are attaching an existing SolidWorks file or starting a new design from scratch.",
        "Choose Workflow -> Configure -> Inspect/Clarify -> Plan -> Execute",
    )