solidworks_mcp.ui.services.session_service¶

solidworks_mcp.ui.services.session_service ¶

Session management and dashboard state assembly for the Prefab CAD dashboard.

Responsibilities (Single Responsibility principle): - Create and hydrate dashboard session rows. - Persist user decisions (brief approval, family acceptance, workflow selection, preferences, notes, context snapshots). - Assemble the DashboardUIState payload consumed by the Prefab UI renderer.

Does NOT own: adapter calls, LLM calls, preview export, RAG ingestion.

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_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'

_DEFAULT_CHECKPOINT_SPECS `module-attribute` ¶

_DEFAULT_CHECKPOINT_SPECS: list[dict[str, Any]] = [{'title': 'Base profile', 'goal': 'Create the base sketch profile', 'tools': ['create_sketch', 'add_line'], 'rationale': 'Establish the primary profile before any 3D feature.'}, {'title': 'Extrude body', 'goal': 'Create the main body', 'tools': ['create_extrusion'], 'rationale': 'Turn the approved sketch into the primary solid.'}, {'title': 'Hole pattern', 'goal': 'Add fastener holes and cable clearance', 'tools': ['create_sketch', 'create_cut'], 'rationale': 'Apply mounting features after the body dimensions are stable.'}, {'title': 'Clearance verify', 'goal': 'Check fit and interference', 'tools': ['check_interference'], 'rationale': 'Validate the assembly path before release or print export.'}]

_META_NAMES `module-attribute` ¶

_META_NAMES = frozenset({'sensors', 'annotations', 'history', 'design binder', 'solid bodies', 'surface bodies', 'lights, cameras and scene', 'equations', 'favorites', 'selection sets', '3d views'})

_META_TYPES `module-attribute` ¶

_META_TYPES = frozenset({'sensorfolder', 'annotationfolder', 'historyfolder', 'designbinder', 'solidbodyfolder', 'surfacebodyfolder', 'lightsfolder', 'mategroup'})

_UJOINT_CHECKPOINT_SPECS `module-attribute` ¶

_UJOINT_CHECKPOINT_SPECS: list[dict[str, Any]] = [{'title': 'Reference mode selection', 'goal': 'Choose target mode (reference reproduction or print variant) and inspect constraints', 'tools': ['open_model', 'list_features', 'create_sketch', 'create_extrusion', 'create_cut'], 'rationale': 'Locks feature-order and dimension guardrails before generating the rest of the set.'}, {'title': 'Part set planning', 'goal': 'Generate the remaining U-joint part plans (yokes, spider, pin, crank family)', 'tools': ['classify_feature_tree', 'create_part', 'create_sketch', 'create_extrusion', 'create_revolve'], 'rationale': 'Decomposes U-joint into deterministic part-level steps with print-aware constraints.'}, {'title': 'Assembly build', 'goal': 'Create UJoint.SLDASM and add deterministic mates with interference checks', 'tools': ['create_assembly', 'check_interference'], 'rationale': 'Ensures mate strategy and collision checks are covered before QA.'}, {'title': 'Final quality checks', 'goal': 'Run final pass/fail checks and export preview images', 'tools': ['list_features', 'get_model_info', 'export_image'], 'rationale': 'Confirms required feature trees and assembly readiness for release.'}]

Classes¶

DashboardCheckpoint ¶

Bases: BaseModel

One checkpoint row shown in the planning/status area.

Attributes:

Name	Type	Description
`goal`	`str`	The goal value.
`model_config`	`Any`	The model config value.
`status`	`str`	The status value.
`step`	`str`	The step value.
`tools`	`str`	The tools value.

DashboardEvidenceRow ¶

Bases: BaseModel

One evidence row shown in retrieval context.

Attributes:

Name	Type	Description
`detail`	`str`	The detail value.
`model_config`	`Any`	The model config value.
`score`	`str`	The score value.
`source`	`str`	The source value.

DashboardUIState ¶

Bases: BaseModel

Top-level dashboard state shared between backend and Prefab UI.

Field descriptions are user-facing semantics and validation intent.

Attributes:

Name	Type	Description
`accepted_family`	`str`	The accepted family value.
`active_model_configuration`	`str`	The active model configuration value.
`active_model_path`	`str`	The active model path value.
`active_model_status`	`str`	The active model status value.
`active_model_type`	`str`	The active model type value.
`api_origin`	`str`	The api origin value.
`assumptions_text`	`str`	The assumptions text value.
`canonical_prompt_text`	`str`	The canonical prompt text value.
`checkpoints`	`list[DashboardCheckpoint]`	The checkpoints value.
`checkpoints_text`	`str`	The checkpoints text value.
`clarifying_questions_text`	`str`	The clarifying questions text value.
`context_file_input`	`str`	The context file input value.
`context_load_status`	`str`	The context load status value.
`context_name_input`	`str`	The context name input value.
`context_save_status`	`str`	The context save status value.
`context_text`	`str`	The context text value.
`context_used_pct`	`int`	The context used pct value.
`ctx_tick`	`int`	The ctx tick value.
`docs_context_text`	`str`	The docs context text value.
`docs_query`	`str`	The docs query value.
`evidence_rows`	`list[DashboardEvidenceRow]`	The evidence rows value.
`evidence_rows_text`	`str`	The evidence rows text value.
`family_confidence`	`str`	The family confidence value.
`family_evidence_text`	`str`	The family evidence text value.
`family_warning_text`	`str`	The family warning text value.
`feature_grounding_warning_text`	`str`	The feature grounding warning text value.
`feature_target_status`	`str`	The feature target status value.
`feature_target_text`	`str`	The feature target text value.
`feature_tree_items`	`list[dict]`	The feature tree items value.
`flow_header_text`	`str`	The flow header text value.
`latest_error_text`	`str`	The latest error text value.
`latest_message`	`str`	The latest message value.
`latest_tool`	`str`	The latest tool value.
`local_endpoint`	`str`	The local endpoint value.
`local_model_available`	`bool`	The local model available value.
`local_model_busy`	`bool`	The local model busy value.
`local_model_label`	`str`	The local model label value.
`local_model_pull_command`	`str`	The local model pull command value.
`local_model_recommended_ollama_model`	`str`	The local model recommended ollama model value.
`local_model_recommended_tier`	`str`	The local model recommended tier value.
`local_model_status_text`	`str`	The local model status text value.
`manual_sync_ready`	`bool`	The manual sync ready value.
`mocked_tools_text`	`str`	The mocked tools text value.
`model_config`	`Any`	The model config value.
`model_context_text`	`str`	The model context text value.
`model_name`	`str`	The model name value.
`model_path_input_chooser`	`str`	The model path input chooser value.
`model_path_input_edit`	`str`	The model path input edit value.
`model_profile`	`str`	The model profile value.
`model_provider`	`str`	The model provider value.
`normalized_brief`	`str`	The normalized brief value.
`notes_text`	`str`	The notes text value.
`orchestration_status`	`str`	The orchestration status value.
`preview_orientation`	`str`	The preview orientation value.
`preview_status`	`str`	The preview status value.
`preview_url`	`str`	The preview url value.
`preview_view_urls`	`dict[str, str]`	The preview view urls value.
`preview_viewer_url`	`str`	The preview viewer url value.
`proposed_family`	`str`	The proposed family value.
`rag_chunk_count`	`int`	The rag chunk count value.
`rag_index_path`	`str`	The rag index path value.
`rag_namespace`	`str`	The rag namespace value.
`rag_provenance_text`	`str`	The rag provenance text value.
`rag_source_path`	`str`	The rag source path value.
`rag_status`	`str`	The rag status value.
`readiness_adapter_mode`	`str`	The readiness adapter mode value.
`readiness_db_ready`	`bool`	The readiness db ready value.
`readiness_preview_ready`	`bool`	The readiness preview ready value.
`readiness_provider_configured`	`bool`	The readiness provider configured value.
`readiness_summary`	`str`	The readiness summary value.
`remediation_hint`	`str`	The remediation hint value.
`selected_feature_name`	`str`	The selected feature name value.
`session_id`	`str`	The session id value.
`structured_rendering_enabled`	`bool`	The structured rendering enabled value.
`tool_history_text`	`str`	The tool history text value.
`uploaded_file_payloads`	`list[dict[str, Any]]`	The uploaded file payloads value.
`user_clarification_answer`	`str`	The user clarification answer value.
`user_goal`	`str`	The user goal value.
`workflow_guidance_text`	`str`	The workflow guidance text value.
`workflow_label`	`str`	The workflow label value.
`workflow_mode`	`str`	The workflow mode value.

Functions¶

_build_checkpoint_rows ¶

_build_checkpoint_rows(session_id: str, *, db_path: Path | None, is_new_design_clean: bool) -> list[dict[str, Any]]

Assemble the checkpoint list for the dashboard state payload.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	Dashboard session identifier.	required
`db_path`	`Path \| None`	Optional SQLite override.	required
`is_new_design_clean`	`bool`	Whether this is a fresh new-design session.	required

Returns:

Type	Description
`list[dict[str, Any]]`	List of :class:`DashboardCheckpoint` dicts.

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

def _build_checkpoint_rows(
    session_id: str,
    *,
    db_path: Path | None,
    is_new_design_clean: bool,
) -> list[dict[str, Any]]:
    """Assemble the checkpoint list for the dashboard state payload.

    Args:
        session_id: Dashboard session identifier.
        db_path: Optional SQLite override.
        is_new_design_clean: Whether this is a fresh new-design session.

    Returns:
        List of :class:`DashboardCheckpoint` dicts.
    """
    checkpoints: list[dict[str, Any]] = []
    for row in list_plan_checkpoints(session_id, db_path=db_path):
        planned = parse_json_blob(row["planned_action_json"])
        result_payload = parse_json_blob(row.get("result_json"))
        if result_payload.get("status") == "error":
            status = "failed"
        elif row["executed"]:
            status = "executed"
        elif row["approved_by_user"]:
            status = "approved"
        else:
            status = "queued"

        if is_new_design_clean and not row["executed"]:
            status = "queued"

        mocked_tools = result_payload.get("mocked_tools", [])
        tools_text = ", ".join(_planned_tools(planned))
        if mocked_tools:
            tools_text = f"{tools_text} [MOCKED: {', '.join(mocked_tools)}]"

        checkpoints.append(
            DashboardCheckpoint(
                step=str(row["checkpoint_index"]),
                goal=planned.get("goal") or row["title"],
                tools=tools_text,
                status=status,
            ).model_dump()
        )
    return checkpoints

_build_evidence_rows ¶

_build_evidence_rows(session_id: str, *, db_path: Path | None, active_model_path: str, is_new_design_clean: bool) -> list[dict[str, Any]]

Assemble the filtered evidence-link rows for the dashboard state payload.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	Dashboard session identifier.	required
`db_path`	`Path \| None`	Optional SQLite override.	required
`active_model_path`	`str`	Currently attached model path.	required
`is_new_design_clean`	`bool`	Whether this is a fresh new-design session.	required

Returns:

Type	Description
`list[dict[str, Any]]`	List of :class:`DashboardEvidenceRow` dicts.

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

def _build_evidence_rows(
    session_id: str,
    *,
    db_path: Path | None,
    active_model_path: str,
    is_new_design_clean: bool,
) -> list[dict[str, Any]]:
    """Assemble the filtered evidence-link rows for the dashboard state payload.

    Args:
        session_id: Dashboard session identifier.
        db_path: Optional SQLite override.
        active_model_path: Currently attached model path.
        is_new_design_clean: Whether this is a fresh new-design session.

    Returns:
        List of :class:`DashboardEvidenceRow` dicts.
    """
    if is_new_design_clean:
        return []

    all_evidence = list_evidence_links(session_id, db_path=db_path)
    model_scoped_sources = {"active_model", "feature_target"}

    filtered_evidence: list[dict[str, Any]] = []
    for evidence in all_evidence:
        source_type = str(evidence.get("source_type") or "")
        source_id = str(evidence.get("source_id") or "")
        if (
            source_type in model_scoped_sources
            and active_model_path
            and source_id
            and source_id != active_model_path
        ):
            continue
        filtered_evidence.append(evidence)

    # Collapse feature-target rows to show only the latest so the table reflects
    # the current target rather than stale historical misses.
    latest_feature_target: dict[str, Any] | None = None
    compact_evidence: list[dict[str, Any]] = []
    for evidence in filtered_evidence:
        if str(evidence.get("source_type") or "") == "feature_target":
            latest_feature_target = evidence
            continue
        compact_evidence.append(evidence)
    if latest_feature_target is not None:
        compact_evidence.append(latest_feature_target)

    rows: list[dict[str, Any]] = []
    for evidence in compact_evidence[-6:]:
        rows.append(
            DashboardEvidenceRow(
                source=evidence["source_type"],
                detail=evidence["rationale"] or evidence["source_id"],
                score=(
                    f"{evidence['relevance_score']:.2f}"
                    if evidence["relevance_score"] is not None
                    else "-"
                ),
            ).model_dump()
        )
    return rows

_build_feature_tree ¶

_build_feature_tree(session_id: str, *, db_path: Path | None, is_new_design_clean: bool, selected_feature_name: str) -> list[dict[str, Any]]

Load and filter the feature tree from the most recent snapshot.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	Dashboard session identifier.	required
`db_path`	`Path \| None`	Optional SQLite override.	required
`is_new_design_clean`	`bool`	When `True`, returns an empty list.	required
`selected_feature_name`	`str`	Currently selected feature (adds `_selected` marker).	required

Returns:

Type	Description
`list[dict[str, Any]]`	List of feature tree row dicts.

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

def _build_feature_tree(
    session_id: str,
    *,
    db_path: Path | None,
    is_new_design_clean: bool,
    selected_feature_name: str,
) -> list[dict[str, Any]]:
    """Load and filter the feature tree from the most recent snapshot.

    Args:
        session_id: Dashboard session identifier.
        db_path: Optional SQLite override.
        is_new_design_clean: When ``True``, returns an empty list.
        selected_feature_name: Currently selected feature (adds ``_selected`` marker).

    Returns:
        List of feature tree row dicts.
    """
    if is_new_design_clean:
        return []

    feature_tree_items: list[dict[str, Any]] = []
    for snap in list_model_state_snapshots(session_id, db_path=db_path):
        raw_tree = snap.get("feature_tree_json")
        if raw_tree:
            try:
                parsed = json.loads(raw_tree)
                if isinstance(parsed, list):
                    feature_tree_items = [
                        f
                        for f in parsed
                        if f.get("name", "").lower() not in _META_NAMES
                        and f.get("type", "").lower() not in _META_TYPES
                    ]
                    break
            except Exception:
                pass

    if selected_feature_name:
        feature_tree_items = [
            {**f, "_selected": "●" if f.get("name") == selected_feature_name else ""}
            for f in feature_tree_items
        ]
    return feature_tree_items

_checkpoint_specs_for_goal ¶

_checkpoint_specs_for_goal(user_goal: str) -> list[dict[str, Any]]

Select initial checkpoint template by goal text.

Parameters:

Name	Type	Description	Default
`user_goal`	`str`	Goal text provided by the user.	required

Returns:

Type	Description
`list[dict[str, Any]]`	List of checkpoint specification dicts.

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

def _checkpoint_specs_for_goal(user_goal: str) -> list[dict[str, Any]]:
    """Select initial checkpoint template by goal text.

    Args:
        user_goal: Goal text provided by the user.

    Returns:
        List of checkpoint specification dicts.
    """
    goal = (user_goal or "").strip().lower()
    if "u-joint" in goal or "ujoint" in goal:
        return list(_UJOINT_CHECKPOINT_SPECS)
    return list(_DEFAULT_CHECKPOINT_SPECS)

_compute_readiness ¶

_compute_readiness(metadata: dict[str, Any], *, db_ready: bool) -> dict[str, Any]

Compute the four readiness signals used by the status badge row.

Parameters:

Name	Type	Description	Default
`metadata`	`dict[str, Any]`	Session metadata dict.	required
`db_ready`	`bool`	Whether the session database row was found.	required

Returns:

Type	Description
`dict[str, Any]`	Dict with `readiness_*` keys consumed by `DashboardUIState`.

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

def _compute_readiness(
    metadata: dict[str, Any],
    *,
    db_ready: bool,
) -> dict[str, Any]:
    """Compute the four readiness signals used by the status badge row.

    Args:
        metadata: Session metadata dict.
        db_ready: Whether the session database row was found.

    Returns:
        Dict with ``readiness_*`` keys consumed by ``DashboardUIState``.
    """
    import os  # local import

    provider = sanitize_ui_text(metadata.get("model_provider"), "").lower()
    model_name = normalize_model_name_for_provider(
        metadata.get("model_name"),
        provider=provider or None,
        profile=sanitize_ui_text(metadata.get("model_profile"), "balanced"),
    )
    local_endpoint = sanitize_ui_text(
        metadata.get("local_endpoint"),
        os.getenv("SOLIDWORKS_UI_LOCAL_ENDPOINT", "http://127.0.0.1:11434/v1"),
    )

    provider_configured = provider_has_credentials(model_name, local_endpoint)

    try:
        config = load_config()
        adapter_mode = str(getattr(config.adapter_type, "value", config.adapter_type))
    except Exception:
        adapter_mode = "unknown"

    preview_dir = ensure_preview_dir()
    import os as _os

    preview_ready = preview_dir.exists() and _os.access(preview_dir, _os.W_OK)

    checks = {
        "provider": provider_configured,
        "adapter": adapter_mode != "unknown",
        "preview": preview_ready,
        "db": db_ready,
    }
    ready_count = sum(1 for v in checks.values() if v)
    summary = (
        f"Readiness {ready_count}/4 | provider={provider_configured} | "
        f"adapter={adapter_mode} | preview={preview_ready} | db={db_ready}"
    )

    return {
        "readiness_provider_configured": provider_configured,
        "readiness_adapter_mode": adapter_mode,
        "readiness_preview_ready": preview_ready,
        "readiness_db_ready": db_ready,
        "readiness_summary": summary,
    }

_planned_tools ¶

_planned_tools(planned: dict[str, Any]) -> list[str]

Extract the tool-name list from a checkpoint planned-action dict.

Parameters:

Name	Type	Description	Default
`planned`	`dict[str, Any]`	Parsed planned-action JSON dict.	required

Returns:

Type	Description
`list[str]`	List of tool name strings.

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

def _planned_tools(planned: dict[str, Any]) -> list[str]:
    """Extract the tool-name list from a checkpoint planned-action dict.

    Args:
        planned: Parsed planned-action JSON dict.

    Returns:
        List of tool name strings.
    """
    tools = planned.get("tools", [])
    return [str(tool) for tool in tools] if isinstance(tools, list) else []

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

context_file_path ¶

context_file_path(session_id: str, *, context_name: str | None = None, context_dir: Path | None = None) -> Path

Return the canonical path for a context snapshot JSON file.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	Session identifier used as the default filename.	required
`context_name`	`str \| None`	Optional override name (will be slugified).	`None`
`context_dir`	`Path \| None`	Override directory for context files.	`None`

Returns:

Type	Description
`Path`	`Path` pointing at `<context_dir>/<safe_name>.json`.

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

def context_file_path(
    session_id: str,
    *,
    context_name: str | None = None,
    context_dir: Path | None = None,
) -> Path:
    """Return the canonical path for a context snapshot JSON file.

    Args:
        session_id: Session identifier used as the default filename.
        context_name: Optional override name (will be slugified).
        context_dir: Override directory for context files.

    Returns:
        ``Path`` pointing at ``<context_dir>/<safe_name>.json``.
    """
    target_dir = ensure_context_dir(context_dir)
    safe_name = safe_context_name(context_name, session_id)
    return target_dir / f"{safe_name}.json"

default_checkpoint_specs ¶

default_checkpoint_specs() -> list[dict[str, Any]]

Return the four default checkpoint specs used when no LLM plan exists.

Returns:

Type	Description
`list[dict[str, Any]]`	List of checkpoint specification dicts.

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

def default_checkpoint_specs() -> list[dict[str, Any]]:
    """Return the four default checkpoint specs used when no LLM plan exists.

    Returns:
        List of checkpoint specification dicts.
    """
    return list(_DEFAULT_CHECKPOINT_SPECS)

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

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."
    )

get_design_session ¶

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

Return one session row as a dictionary.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	The session id value.	required
`db_path`	`Path \| None`	The db path value. Defaults to None.	`None`

Returns:

Type	Description
`dict[str, Any] \| None`	dict[str, Any] \| None: A dictionary containing the resulting values.

Source code in src/solidworks_mcp/agents/history_db.py

def get_design_session(
    session_id: str, db_path: Path | None = None
) -> dict[str, Any] | None:
    """Return one session row as a dictionary.

    Args:
        session_id (str): The session id value.
        db_path (Path | None): The db path value. Defaults to None.

    Returns:
        dict[str, Any] | None: A dictionary containing the resulting values.
    """
    resolved = init_db(db_path)
    engine = _build_engine(resolved)
    with Session(engine) as session:
        row = session.exec(
            select(DesignSession).where(DesignSession.session_id == session_id)
        ).first()

    if row is None:
        return None
    return {
        "session_id": row.session_id,
        "user_goal": row.user_goal,
        "source_mode": row.source_mode,
        "accepted_family": row.accepted_family,
        "status": row.status,
        "current_checkpoint_index": row.current_checkpoint_index,
        "metadata_json": row.metadata_json,
        "created_at": row.created_at,
        "updated_at": row.updated_at,
    }

insert_plan_checkpoint ¶

insert_plan_checkpoint(*, session_id: str, checkpoint_index: int, title: str, planned_action_json: str, approved_by_user: bool = False, executed: bool = False, result_json: str | None = None, rollback_snapshot_id: int | None = None, db_path: Path | None = None) -> int

Insert a new checkpoint and return its ID.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	The session id value.	required
`checkpoint_index`	`int`	The checkpoint index value.	required
`title`	`str`	The title value.	required
`planned_action_json`	`str`	The planned action json value.	required
`approved_by_user`	`bool`	The approved by user value. Defaults to False.	`False`
`executed`	`bool`	The executed value. Defaults to False.	`False`
`result_json`	`str \| None`	The result json value. Defaults to None.	`None`
`rollback_snapshot_id`	`int \| None`	The rollback snapshot id value. Defaults to None.	`None`
`db_path`	`Path \| None`	The db path value. Defaults to None.	`None`

Returns:

Name	Type	Description
`int`	`int`	The computed numeric result.

Source code in src/solidworks_mcp/agents/history_db.py

def insert_plan_checkpoint(
    *,
    session_id: str,
    checkpoint_index: int,
    title: str,
    planned_action_json: str,
    approved_by_user: bool = False,
    executed: bool = False,
    result_json: str | None = None,
    rollback_snapshot_id: int | None = None,
    db_path: Path | None = None,
) -> int:
    """Insert a new checkpoint and return its ID.

    Args:
        session_id (str): The session id value.
        checkpoint_index (int): The checkpoint index value.
        title (str): The title value.
        planned_action_json (str): The planned action json value.
        approved_by_user (bool): The approved by user value. Defaults to False.
        executed (bool): The executed value. Defaults to False.
        result_json (str | None): The result json value. Defaults to None.
        rollback_snapshot_id (int | None): The rollback snapshot id value. Defaults to None.
        db_path (Path | None): The db path value. Defaults to None.

    Returns:
        int: The computed numeric result.
    """
    resolved = init_db(db_path)
    engine = _build_engine(resolved)
    now = _utc_now_iso()
    with Session(engine) as session:
        row = PlanCheckpoint(
            session_id=session_id,
            checkpoint_index=checkpoint_index,
            title=title,
            planned_action_json=planned_action_json,
            approved_by_user=approved_by_user,
            executed=executed,
            result_json=result_json,
            rollback_snapshot_id=rollback_snapshot_id,
            created_at=now,
            updated_at=now,
        )
        session.add(row)
        session.commit()
        session.refresh(row)
        return int(row.id or 0)

list_evidence_links ¶

list_evidence_links(session_id: str, checkpoint_id: int | None = None, db_path: Path | None = None) -> list[dict[str, Any]]

List evidence rows for a session and optional checkpoint.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	The session id value.	required
`checkpoint_id`	`int \| None`	The checkpoint id value. Defaults to None.	`None`
`db_path`	`Path \| None`	The db path value. Defaults to None.	`None`

Returns:

Type	Description
`list[dict[str, Any]]`	list[dict[str, Any]]: A list containing the resulting items.

Source code in src/solidworks_mcp/agents/history_db.py

def list_evidence_links(
    session_id: str,
    checkpoint_id: int | None = None,
    db_path: Path | None = None,
) -> list[dict[str, Any]]:
    """List evidence rows for a session and optional checkpoint.

    Args:
        session_id (str): The session id value.
        checkpoint_id (int | None): The checkpoint id value. Defaults to None.
        db_path (Path | None): The db path value. Defaults to None.

    Returns:
        list[dict[str, Any]]: A list containing the resulting items.
    """
    resolved = init_db(db_path)
    engine = _build_engine(resolved)
    with Session(engine) as session:
        query = select(EvidenceLink).where(EvidenceLink.session_id == session_id)
        if checkpoint_id is not None:
            query = query.where(EvidenceLink.checkpoint_id == checkpoint_id)
        rows = session.exec(query.order_by(EvidenceLink.id.asc())).all()  # type: ignore[union-attr]

    return [
        {
            "id": row.id,
            "session_id": row.session_id,
            "checkpoint_id": row.checkpoint_id,
            "source_type": row.source_type,
            "source_id": row.source_id,
            "relevance_score": row.relevance_score,
            "rationale": row.rationale,
            "payload_json": row.payload_json,
            "created_at": row.created_at,
        }
        for row in rows
    ]

list_model_state_snapshots ¶

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

List snapshots for a session newest first for diff/rollback flows.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	The session id value.	required
`db_path`	`Path \| None`	The db path value. Defaults to None.	`None`

Returns:

Type	Description
`list[dict[str, Any]]`	list[dict[str, Any]]: A list containing the resulting items.

Source code in src/solidworks_mcp/agents/history_db.py

def list_model_state_snapshots(
    session_id: str,
    db_path: Path | None = None,
) -> list[dict[str, Any]]:
    """List snapshots for a session newest first for diff/rollback flows.

    Args:
        session_id (str): The session id value.
        db_path (Path | None): The db path value. Defaults to None.

    Returns:
        list[dict[str, Any]]: A list containing the resulting items.
    """
    resolved = init_db(db_path)
    engine = _build_engine(resolved)
    with Session(engine) as session:
        rows = session.exec(
            select(ModelStateSnapshot)
            .where(ModelStateSnapshot.session_id == session_id)
            .order_by(ModelStateSnapshot.id.desc())  # type: ignore[union-attr]
        ).all()

    return [
        {
            "id": row.id,
            "session_id": row.session_id,
            "checkpoint_id": row.checkpoint_id,
            "model_path": row.model_path,
            "feature_tree_json": row.feature_tree_json,
            "mass_properties_json": row.mass_properties_json,
            "screenshot_path": row.screenshot_path,
            "state_fingerprint": row.state_fingerprint,
            "created_at": row.created_at,
        }
        for row in rows
    ]

list_plan_checkpoints ¶

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

List all checkpoints for a session.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	The session id value.	required
`db_path`	`Path \| None`	The db path value. Defaults to None.	`None`

Returns:

Type	Description
`list[dict[str, Any]]`	list[dict[str, Any]]: A list containing the resulting items.

Source code in src/solidworks_mcp/agents/history_db.py

def list_plan_checkpoints(
    session_id: str,
    db_path: Path | None = None,
) -> list[dict[str, Any]]:
    """List all checkpoints for a session.

    Args:
        session_id (str): The session id value.
        db_path (Path | None): The db path value. Defaults to None.

    Returns:
        list[dict[str, Any]]: A list containing the resulting items.
    """
    resolved = init_db(db_path)
    engine = _build_engine(resolved)
    with Session(engine) as session:
        rows = session.exec(
            select(PlanCheckpoint)
            .where(PlanCheckpoint.session_id == session_id)
            .order_by(PlanCheckpoint.checkpoint_index.asc())  # type: ignore[union-attr]
        ).all()

    return [
        {
            "id": row.id,
            "session_id": row.session_id,
            "checkpoint_index": row.checkpoint_index,
            "title": row.title,
            "planned_action_json": row.planned_action_json,
            "approved_by_user": row.approved_by_user,
            "executed": row.executed,
            "result_json": row.result_json,
            "rollback_snapshot_id": row.rollback_snapshot_id,
            "created_at": row.created_at,
            "updated_at": row.updated_at,
        }
        for row in rows
    ]

list_tool_call_records ¶

list_tool_call_records(session_id: str, checkpoint_id: int | None = None, db_path: Path | None = None) -> list[dict[str, Any]]

List tool call records for a session and optional checkpoint.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	The session id value.	required
`checkpoint_id`	`int \| None`	The checkpoint id value. Defaults to None.	`None`
`db_path`	`Path \| None`	The db path value. Defaults to None.	`None`

Returns:

Type	Description
`list[dict[str, Any]]`	list[dict[str, Any]]: A list containing the resulting items.

Source code in src/solidworks_mcp/agents/history_db.py

def list_tool_call_records(
    session_id: str,
    checkpoint_id: int | None = None,
    db_path: Path | None = None,
) -> list[dict[str, Any]]:
    """List tool call records for a session and optional checkpoint.

    Args:
        session_id (str): The session id value.
        checkpoint_id (int | None): The checkpoint id value. Defaults to None.
        db_path (Path | None): The db path value. Defaults to None.

    Returns:
        list[dict[str, Any]]: A list containing the resulting items.
    """
    resolved = init_db(db_path)
    engine = _build_engine(resolved)
    with Session(engine) as session:
        query = select(ToolCallRecord).where(ToolCallRecord.session_id == session_id)
        if checkpoint_id is not None:
            query = query.where(ToolCallRecord.checkpoint_id == checkpoint_id)
        rows = session.exec(query.order_by(ToolCallRecord.id.asc())).all()  # type: ignore[union-attr]

    return [
        {
            "id": row.id,
            "session_id": row.session_id,
            "checkpoint_id": row.checkpoint_id,
            "run_id": row.run_id,
            "tool_name": row.tool_name,
            "input_json": row.input_json,
            "output_json": row.output_json,
            "success": row.success,
            "latency_ms": row.latency_ms,
            "created_at": row.created_at,
        }
        for row in rows
    ]

load_config ¶

load_config(config_file: str | None = None) -> SolidWorksMCPConfig

Load configuration from file and environment variables.

Parameters:

Name	Type	Description	Default
`config_file`	`str \| None`	The config file value. Defaults to None.	`None`

Returns:

Name	Type	Description
`SolidWorksMCPConfig`	`SolidWorksMCPConfig`	The result produced by the operation.

Source code in src/solidworks_mcp/config.py

def load_config(config_file: str | None = None) -> SolidWorksMCPConfig:
    """Load configuration from file and environment variables.

    Args:
        config_file (str | None): The config file value. Defaults to None.

    Returns:
        SolidWorksMCPConfig: The result produced by the operation.
    """
    if config_file:
        config_path = Path(config_file)
        if config_path.exists() and config_path.suffix.lower() == ".json":
            import json

            with config_path.open("r", encoding="utf-8") as f:
                data = json.load(f)
            return SolidWorksMCPConfig(**data)
        return SolidWorksMCPConfig.from_env(str(config_path))

    return SolidWorksMCPConfig.from_env()

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)

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_model_name_for_provider ¶

normalize_model_name_for_provider(model_name: str | None, *, provider: str | None, profile: str | None = None) -> str

Normalise a free-form model name into a provider-qualified routing string.

Parameters:

Name	Type	Description	Default
`model_name`	`str \| None`	Raw model name from UI state, may be None or unqualified.	required
`provider`	`str \| None`	Explicit provider override (`"github"`, `"local"`, etc.).	required
`profile`	`str \| None`	Profile tier used as fallback when model_name is empty.	`None`

Returns:

Type	Description
`str`	Provider-qualified model name (e.g. `"github:openai/gpt-4.1"`).

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

def normalize_model_name_for_provider(
    model_name: str | None,
    *,
    provider: str | None,
    profile: str | None = None,
) -> str:
    """Normalise a free-form model name into a provider-qualified routing string.

    Args:
        model_name: Raw model name from UI state, may be None or unqualified.
        provider: Explicit provider override (``"github"``, ``"local"``, etc.).
        profile: Profile tier used as fallback when *model_name* is empty.

    Returns:
        Provider-qualified model name (e.g. ``"github:openai/gpt-4.1"``).
    """
    normalized_provider = (provider or "github").strip().lower()
    normalized_profile = (profile or "balanced").strip().lower()
    raw_model = sanitize_ui_text(model_name, "")

    if not raw_model:
        return default_model_for_profile(normalized_provider, normalized_profile)

    # Already provider-qualified.
    if ":" in raw_model:
        return raw_model

    if normalized_provider == "local":
        return f"local:{raw_model}"
    if normalized_provider == "github":
        if "/" in raw_model:
            return f"github:{raw_model}"
        return f"github:openai/{raw_model}"
    if normalized_provider in {"openai", "anthropic"}:
        return f"{normalized_provider}:{raw_model}"

    return raw_model

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

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

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)

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_plan_checkpoint ¶

update_plan_checkpoint(checkpoint_id: int, *, approved_by_user: bool | None = None, executed: bool | None = None, result_json: str | None = None, rollback_snapshot_id: int | None = None, db_path: Path | None = None) -> None

Patch checkpoint approval/execution fields.

Parameters:

Name	Type	Description	Default
`checkpoint_id`	`int`	The checkpoint id value.	required
`approved_by_user`	`bool \| None`	The approved by user value. Defaults to None.	`None`
`executed`	`bool \| None`	The executed value. Defaults to None.	`None`
`result_json`	`str \| None`	The result json value. Defaults to None.	`None`
`rollback_snapshot_id`	`int \| None`	The rollback snapshot id value. Defaults to None.	`None`
`db_path`	`Path \| None`	The db path value. Defaults to None.	`None`

Returns:

Name	Type	Description
`None`	`None`	None.

Source code in src/solidworks_mcp/agents/history_db.py

def update_plan_checkpoint(
    checkpoint_id: int,
    *,
    approved_by_user: bool | None = None,
    executed: bool | None = None,
    result_json: str | None = None,
    rollback_snapshot_id: int | None = None,
    db_path: Path | None = None,
) -> None:
    """Patch checkpoint approval/execution fields.

    Args:
        checkpoint_id (int): The checkpoint id value.
        approved_by_user (bool | None): The approved by user value. Defaults to None.
        executed (bool | None): The executed value. Defaults to None.
        result_json (str | None): The result json value. Defaults to None.
        rollback_snapshot_id (int | None): The rollback snapshot id value. Defaults to None.
        db_path (Path | None): The db path value. Defaults to None.

    Returns:
        None: None.
    """
    resolved = init_db(db_path)
    engine = _build_engine(resolved)
    with Session(engine) as session:
        row = session.exec(
            select(PlanCheckpoint).where(PlanCheckpoint.id == checkpoint_id)
        ).first()
        if row is None:
            return
        if approved_by_user is not None:
            row.approved_by_user = approved_by_user
        if executed is not None:
            row.executed = executed
        if result_json is not None:
            row.result_json = result_json
        if rollback_snapshot_id is not None:
            row.rollback_snapshot_id = rollback_snapshot_id
        row.updated_at = _utc_now_iso()
        session.add(row)
        session.commit()

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)

upsert_design_session ¶

upsert_design_session(*, session_id: str, user_goal: str, source_mode: str = 'model', accepted_family: str | None = None, status: str = 'active', current_checkpoint_index: int = 0, metadata_json: str | None = None, db_path: Path | None = None) -> None

Create or update one interactive design session row.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	The session id value.	required
`user_goal`	`str`	The user goal value.	required
`source_mode`	`str`	The source mode value. Defaults to "model".	`'model'`
`accepted_family`	`str \| None`	The accepted family value. Defaults to None.	`None`
`status`	`str`	The status value. Defaults to "active".	`'active'`
`current_checkpoint_index`	`int`	The current checkpoint index value. Defaults to 0.	`0`
`metadata_json`	`str \| None`	The metadata json value. Defaults to None.	`None`
`db_path`	`Path \| None`	The db path value. Defaults to None.	`None`

Returns:

Name	Type	Description
`None`	`None`	None.

Source code in src/solidworks_mcp/agents/history_db.py

def upsert_design_session(
    *,
    session_id: str,
    user_goal: str,
    source_mode: str = "model",
    accepted_family: str | None = None,
    status: str = "active",
    current_checkpoint_index: int = 0,
    metadata_json: str | None = None,
    db_path: Path | None = None,
) -> None:
    """Create or update one interactive design session row.

    Args:
        session_id (str): The session id value.
        user_goal (str): The user goal value.
        source_mode (str): The source mode value. Defaults to "model".
        accepted_family (str | None): The accepted family value. Defaults to None.
        status (str): The status value. Defaults to "active".
        current_checkpoint_index (int): The current checkpoint index value. Defaults to 0.
        metadata_json (str | None): The metadata json value. Defaults to None.
        db_path (Path | None): The db path value. Defaults to None.

    Returns:
        None: None.
    """
    resolved = init_db(db_path)
    engine = _build_engine(resolved)
    now = _utc_now_iso()
    with Session(engine) as session:
        row = session.exec(
            select(DesignSession).where(DesignSession.session_id == session_id)
        ).first()
        if row is None:
            session.add(
                DesignSession(
                    session_id=session_id,
                    user_goal=user_goal,
                    source_mode=source_mode,
                    accepted_family=accepted_family,
                    status=status,
                    current_checkpoint_index=current_checkpoint_index,
                    metadata_json=metadata_json,
                    created_at=now,
                    updated_at=now,
                )
            )
        else:
            row.user_goal = user_goal
            row.source_mode = source_mode
            row.accepted_family = accepted_family
            row.status = status
            row.current_checkpoint_index = current_checkpoint_index
            row.metadata_json = metadata_json
            row.updated_at = now
            session.add(row)
        session.commit()

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",
    )

solidworks_mcp.ui.services.session_service¶