solidworks_mcp.ui.service

solidworks_mcp.ui.service

Compatibility shim for legacy imports from solidworks_mcp.ui.service.

The UI backend was split into focused modules under solidworks_mcp.ui.services. This module preserves historical import paths used by tests and external callers.

Attributes

DEFAULT_API_ORIGIN module-attribute

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

DEFAULT_CONTEXT_DIR module-attribute

DEFAULT_CONTEXT_DIR = _DEFAULT_CONTEXT_DIR

DEFAULT_PREVIEW_ORIENTATION module-attribute

DEFAULT_PREVIEW_ORIENTATION = 'current'

DEFAULT_RAG_DIR module-attribute

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

DEFAULT_SESSION_ID module-attribute

DEFAULT_SESSION_ID = 'prefab-dashboard'

DEFAULT_SOURCE_MODE module-attribute

DEFAULT_SOURCE_MODE = 'prompt'

DEFAULT_UPLOADED_MODEL_DIR module-attribute

DEFAULT_UPLOADED_MODEL_DIR = _DEFAULT_UPLOADED_MODEL_DIR

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'

RecoverableFailure module-attribute

RecoverableFailure = RecoverableFailure

SUPPORTED_MODEL_UPLOAD_SUFFIXES module-attribute

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

_ORIG_BUILD_AGENT_MODEL module-attribute

_ORIG_BUILD_AGENT_MODEL = _build_agent_model

_ORIG_RUN_STRUCTURED_AGENT module-attribute

_ORIG_RUN_STRUCTURED_AGENT = _run_structured_agent

__all__ module-attribute

__all__ = [*[name for name in (globals()) if not startswith('__')]]

_normalize_feature_targets module-attribute

_normalize_feature_targets = _normalize_feature_targets

_safe_context_name module-attribute

_safe_context_name = _safe_context_name

_sanitize_model_path_text module-attribute

_sanitize_model_path_text = _sanitize_model_path_text

_sanitize_preview_viewer_url module-attribute

_sanitize_preview_viewer_url = _sanitize_preview_viewer_url

Classes

CheckpointCandidate

Bases: BaseModel

One suggested execution checkpoint.

Attributes:

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

ClarificationResponse

Bases: BaseModel

LLM response for goal clarification.

Attributes:

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

FamilyInspection

Bases: BaseModel

LLM response for feature-family classification.

Attributes:

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

_AgentRecoverableFailure

Bases: BaseModel

Typed failure output used when agent needs user-guided retry.

Attributes:

Name	Type	Description
explanation	str	The explanation value.
remediation_steps	list[str]	The remediation steps value.
retry_focus	str | None	The retry focus value.
should_retry	bool	The should retry value.

Functions

_build_agent_model

_build_agent_model(model_name: str, local_endpoint: str | None = None)

Source code in src/solidworks_mcp/ui/service.py

def _build_agent_model(model_name: str, local_endpoint: str | None = None):
    with _temporary_module_bindings(
        _llm_service,
        OpenAIProvider=OpenAIProvider,
        OpenAIChatModel=OpenAIChatModel,
    ):
        return _ORIG_BUILD_AGENT_MODEL(model_name, local_endpoint)

_chunk_text

_chunk_text(text: str, chunk_size: int = 1000, overlap: int = 150) -> list[str]

Split long text into overlapping chunks for simple local retrieval.

Parameters:

Name	Type	Description	Default
text	str	Input text processed by the operation.	required
chunk_size	int	Maximum number of characters to keep in each chunk. Defaults to 1000.	1000
overlap	int	Number of overlapping characters between chunks. Defaults to 150.	150

Returns:

Type	Description
list[str]	list[str]: A list containing the resulting items.

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

def _chunk_text(text: str, chunk_size: int = 1000, overlap: int = 150) -> list[str]:
    """Split long text into overlapping chunks for simple local retrieval.

    Args:
        text (str): Input text processed by the operation.
        chunk_size (int): Maximum number of characters to keep in each chunk. Defaults to
                          1000.
        overlap (int): Number of overlapping characters between chunks. Defaults to 150.

    Returns:
        list[str]: A list containing the resulting items.
    """
    normalized = (text or "").strip()
    if not normalized:
        return []

    if len(normalized) <= chunk_size:
        return [normalized]

    chunks: list[str] = []
    start = 0
    while start < len(normalized):
        end = min(start + chunk_size, len(normalized))
        chunks.append(normalized[start:end])
        if end == len(normalized):
            break
        start = max(0, end - overlap)
    return chunks

_context_file_path

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

Source code in src/solidworks_mcp/ui/service.py

def _context_file_path(
    session_id: str,
    *,
    context_name: str | None = None,
    context_dir: Path | None = None,
) -> Path:
    return _context_file_path_impl(
        session_id,
        context_name=context_name,
        context_dir=context_dir or DEFAULT_CONTEXT_DIR,
    )

_context_file_path_impl

_context_file_path_impl(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_model_for_profile

_default_model_for_profile(provider: str, profile: str) -> str

Source code in src/solidworks_mcp/ui/service.py

def _default_model_for_profile(provider: str, profile: str) -> str:
    if provider == "local":
        profile_models = {
            "small": "local:gemma4:e2b",
            "balanced": "local:gemma4:e4b",
            "large": "local:gemma4:26b",
        }
        return profile_models.get(
            (profile or "balanced").lower(), profile_models["balanced"]
        )

    profile_models = {
        "small": "github:openai/gpt-4.1-mini",
        "balanced": "github:openai/gpt-4.1",
        "large": "github:openai/gpt-4.1",
    }
    return profile_models.get(
        (profile or "balanced").lower(), profile_models["balanced"]
    )

_ensure_provider_credentials

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

Verify that the required provider credential is present, raising otherwise.

Parameters:

Name	Type	Description	Default
model_name	str	Provider-qualified model name (e.g. "github:openai/gpt-4.1").	required
local_endpoint	str | None	Optional base URL for a local LLM server.	None

Raises:

Type	Description
RuntimeError	When the required credential is missing.

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

def _ensure_provider_credentials(
    model_name: str,
    local_endpoint: str | None = None,
) -> None:
    """Verify that the required provider credential is present, raising otherwise.

    Args:
        model_name: Provider-qualified model name (e.g. ``"github:openai/gpt-4.1"``).
        local_endpoint: Optional base URL for a local LLM server.

    Raises:
        RuntimeError: When the required credential is missing.
    """
    # TODO: security review — subprocess token extraction.  Replace with env-only lookup.
    if model_name.startswith("github:"):
        github_token = os.getenv("GITHUB_API_KEY") or os.getenv("GH_TOKEN")
        if not github_token:
            try:
                result = subprocess.run(  # noqa: S603
                    ["gh", "auth", "token"],  # noqa: S607
                    capture_output=True,
                    text=True,
                    timeout=5,
                    check=False,
                )
            except Exception:
                result = None  # type: ignore[assignment]
            if result and result.returncode == 0:
                github_token = result.stdout.strip()
        if not github_token:
            raise RuntimeError(
                "Set GH_TOKEN or GITHUB_API_KEY with models:read scope before using the dashboard LLM actions."
            )
        os.environ.setdefault("GITHUB_API_KEY", github_token)
        return

    if model_name.startswith("openai:") and not os.getenv("OPENAI_API_KEY"):
        raise RuntimeError("Set OPENAI_API_KEY before using OpenAI model routing.")

    if model_name.startswith("anthropic:") and not os.getenv("ANTHROPIC_API_KEY"):
        raise RuntimeError(
            "Set ANTHROPIC_API_KEY before using Anthropic model routing."
        )

    if model_name.startswith("local:"):
        # Endpoint defaults to http://127.0.0.1:11434/v1 inside _build_agent_model;
        # no credential is needed for a local Ollama server.
        pass

_feature_grounding_warning_text

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

Return a warning string when feature grounding cannot proceed.

Parameters:

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

Returns:

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

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

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

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

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

_feature_target_status

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

Compute matched / missing feature target status.

Parameters:

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

Returns:

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

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

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

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

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

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

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

_filter_docs_text

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

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

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

Parameters:

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

Returns:

Type	Description
str	Filtered and truncated text snippet.

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

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

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

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

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

_looks_like_path_token

_looks_like_path_token(token: str) -> bool

Return True if token looks like a filesystem path rather than a feature name.

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

def _looks_like_path_token(token: str) -> bool:
    """Return ``True`` if *token* looks like a filesystem path rather than a feature name."""
    normalized = token.strip()
    if not normalized:
        return False
    lowered = normalized.lower()
    if len(normalized) >= 2 and normalized[1] == ":" and normalized[0].isalpha():
        return True
    if "\\" in normalized or "/" in normalized:
        return True
    if lowered.endswith(
        (
            ".sldprt",
            ".sldasm",
            ".slddrw",
            ".step",
            ".stp",
            ".iges",
            ".igs",
            ".x_t",
            ".x_b",
        )
    ):
        return True
    return False

_materialize_uploaded_model

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

Source code in src/solidworks_mcp/ui/service.py

def _materialize_uploaded_model(
    session_id: str,
    uploaded_files: list[dict[str, Any]] | None,
    *,
    upload_dir: Path | None = None,
) -> Path:
    return _materialize_uploaded_model_impl(
        session_id,
        uploaded_files,
        upload_dir=upload_dir or DEFAULT_UPLOADED_MODEL_DIR,
    )

_materialize_uploaded_model_impl

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

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

Parameters:

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

Returns:

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

Raises:

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

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

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

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

    Returns:
        Path pointing at the decoded file on disk.

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

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

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

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

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

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

_merge_metadata

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

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

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

Parameters:

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

Returns:

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

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

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

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

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

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

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

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

_normalize_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

_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 {}

_provider_from_model_name

_provider_from_model_name(model_name: str) -> str

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

Parameters:

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

Returns:

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

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

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

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

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

_provider_has_credentials

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

Check whether the required credentials exist for the given model.

Parameters:

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

Returns:

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

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

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

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

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

_read_reference_source

_read_reference_source(source_path: Path) -> str

Source code in src/solidworks_mcp/ui/service.py

def _read_reference_source(source_path: Path) -> str:
    return _utils_service.read_reference_source(source_path)

_read_reference_url

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

Source code in src/solidworks_mcp/ui/service.py

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

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

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

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

    return decoded.strip(), label

_run_checkpoint_tools async

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

Source code in src/solidworks_mcp/ui/service.py

async def _run_checkpoint_tools(planned: dict[str, Any]) -> dict[str, Any]:
    with _temporary_module_bindings(
        _checkpoint_service,
        create_adapter=create_adapter,
        load_config=load_config,
    ):
        return await _checkpoint_service._run_checkpoint_tools(planned)

_run_structured_agent async

_run_structured_agent(*args: Any, **kwargs: Any)

Source code in src/solidworks_mcp/ui/service.py

async def _run_structured_agent(*args: Any, **kwargs: Any):
    with _temporary_module_bindings(
        _llm_service,
        Agent=Agent,
        RecoverableFailure=RecoverableFailure,
        _ensure_provider_credentials=_ensure_provider_credentials,
        _build_agent_model=_build_agent_model,
    ):
        result = await _ORIG_RUN_STRUCTURED_AGENT(*args, **kwargs)
        if isinstance(result, RecoverableFailure):
            return result
        if all(
            hasattr(result, name)
            for name in (
                "explanation",
                "remediation_steps",
                "retry_focus",
                "should_retry",
            )
        ):
            return RecoverableFailure(
                explanation=str(getattr(result, "explanation", "")),
                remediation_steps=list(getattr(result, "remediation_steps", []) or []),
                retry_focus=str(getattr(result, "retry_focus", "")) or None,
                should_retry=bool(getattr(result, "should_retry", False)),
            )
        return result

_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

_temporary_module_bindings

_temporary_module_bindings(module: Any, **bindings: Any)

Source code in src/solidworks_mcp/ui/service.py

@contextmanager
def _temporary_module_bindings(module: Any, **bindings: Any):
    original: dict[str, Any] = {}
    for name, value in bindings.items():
        if hasattr(module, name):
            original[name] = getattr(module, name)
        setattr(module, name, value)
    try:
        yield
    finally:
        for name in bindings:
            if name in original:
                setattr(module, name, original[name])

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

_trace_json_default(value: Any) -> str

JSON serialisation default: convert unknown types to their str repr.

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

def _trace_json_default(value: Any) -> str:
    """JSON serialisation default: convert unknown types to their str repr."""
    return str(value)

_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

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

accept_family_choice

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

Accept the proposed family classification and advance session status.

Parameters:

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

Returns:

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

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

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

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

    Returns:
        Full dashboard state payload.
    """
    session_row = ensure_dashboard_session(session_id, db_path=db_path)
    metadata = parse_json_blob(session_row.get("metadata_json"))
    accepted_family = family or metadata.get("proposed_family") or "unknown"
    upsert_design_session(
        session_id=session_id,
        user_goal=session_row.get("user_goal") or DEFAULT_USER_GOAL,
        source_mode=session_row.get("source_mode") or DEFAULT_SOURCE_MODE,
        accepted_family=accepted_family,
        status="planned",
        current_checkpoint_index=session_row.get("current_checkpoint_index") or 0,
        metadata_json=session_row.get("metadata_json"),
        db_path=db_path,
    )
    persist_ui_action(
        session_id,
        tool_name="ui.accept_family",
        db_path=db_path,
        metadata_updates={
            "accepted_family": accepted_family,
            "latest_message": f"Family accepted: {accepted_family}.",
            "latest_error_text": "",
            "remediation_hint": "",
        },
        input_payload={"family": accepted_family},
    )
    return build_dashboard_state(session_id, db_path=db_path)

approve_design_brief

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

Persist the accepted design goal and return updated dashboard state.

Parameters:

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

Returns:

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

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

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

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

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

build_dashboard_state

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

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

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

Parameters:

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

Returns:

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

build_dashboard_trace_payload

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

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

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

Parameters:

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

Returns:

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

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

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

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

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

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

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

connect_target_model async

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

Source code in src/solidworks_mcp/ui/service.py

async def connect_target_model(
    session_id: str,
    *,
    model_path: str | None = None,
    uploaded_files: list[dict[str, Any]] | None = None,
    feature_target_text: str = "",
    db_path: Path | None = None,
    api_origin: str = DEFAULT_API_ORIGIN,
) -> dict[str, Any]:
    with (
        _temporary_module_bindings(
            _model_service,
            create_adapter=create_adapter,
            load_config=load_config,
            refresh_preview=refresh_preview,
        ),
        _temporary_module_bindings(
            _preview_service,
            create_adapter=create_adapter,
            load_config=load_config,
        ),
    ):
        return await _model_service.connect_target_model(
            session_id,
            model_path=model_path,
            uploaded_files=uploaded_files,
            feature_target_text=feature_target_text,
            db_path=db_path,
            api_origin=api_origin,
        )

create_adapter async

create_adapter(config: SolidWorksMCPConfig) -> SolidWorksAdapter

Async factory function for creating SolidWorks adapters.

Parameters:

Name	Type	Description	Default
config	SolidWorksMCPConfig	Configuration values for the operation.	required

Returns:

Name	Type	Description
SolidWorksAdapter	SolidWorksAdapter	The result produced by the operation.

Source code in src/solidworks_mcp/adapters/factory.py

async def create_adapter(config: SolidWorksMCPConfig) -> SolidWorksAdapter:
    """Async factory function for creating SolidWorks adapters.

    Args:
        config (SolidWorksMCPConfig): Configuration values for the operation.

    Returns:
        SolidWorksAdapter: The result produced by the operation.
    """
    # Register adapters if not already done
    _register_default_adapters()

    # Create adapter using factory
    adapter = AdapterFactory.create_adapter(config)

    return adapter

ensure_context_dir

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

Source code in src/solidworks_mcp/ui/service.py

def ensure_context_dir(context_dir: Path | None = None) -> Path:
    return _utils_service.ensure_context_dir(context_dir or DEFAULT_CONTEXT_DIR)

ensure_dashboard_session

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

Ensure one dashboard session row and default checkpoints exist.

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

Parameters:

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

Returns:

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

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

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

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

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

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

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

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

ensure_preview_dir

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

Create and return the preview image directory.

Parameters:

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

Returns:

Type	Description
Path	Resolved Path that is guaranteed to exist.

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

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

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

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

ensure_uploaded_model_dir

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

Source code in src/solidworks_mcp/ui/service.py

def ensure_uploaded_model_dir(upload_dir: Path | None = None) -> Path:
    return _utils_service.ensure_uploaded_model_dir(
        upload_dir or DEFAULT_UPLOADED_MODEL_DIR
    )

execute_next_checkpoint async

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

Source code in src/solidworks_mcp/ui/service.py

async def execute_next_checkpoint(
    session_id: str,
    *,
    db_path: Path | None = None,
) -> dict[str, Any]:
    with (
        _temporary_module_bindings(
            _session_service,
            ensure_dashboard_session=ensure_dashboard_session,
            build_dashboard_state=build_dashboard_state,
        ),
        _temporary_module_bindings(
            _checkpoint_service,
            list_plan_checkpoints=list_plan_checkpoints,
            merge_metadata=_merge_metadata,
        ),
    ):
        return await _checkpoint_service.execute_next_checkpoint(
            session_id, db_path=db_path
        )

feature_grounding_warning_text

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

Return a warning string when feature grounding cannot proceed.

Parameters:

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

Returns:

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

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

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

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

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

feature_target_status

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

Compute matched / missing feature target status.

Parameters:

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

Returns:

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

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

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

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

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

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

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

fetch_docs_context

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

Source code in src/solidworks_mcp/ui/service.py

def fetch_docs_context(
    session_id: str,
    *,
    docs_query: str = "",
    db_path: Path | None = None,
    api_origin: str = DEFAULT_API_ORIGIN,
) -> dict[str, Any]:
    with _temporary_module_bindings(
        _docs_service,
        urlopen=urlopen,
    ):
        return _docs_service.fetch_docs_context(
            session_id,
            docs_query=docs_query,
            db_path=db_path,
            api_origin=api_origin,
        )

filter_docs_text

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

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

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

Parameters:

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

Returns:

Type	Description
str	Filtered and truncated text snippet.

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

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

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

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

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

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,
    }

highlight_feature async

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

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

Parameters:

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

Returns:

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

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

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

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

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

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

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

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

ingest_reference_source

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

Source code in src/solidworks_mcp/ui/service.py

def ingest_reference_source(
    session_id: str,
    *,
    source_path: str,
    namespace: str,
    chunk_size: int = 1200,
    overlap: int = 200,
    db_path: Path | None = None,
) -> dict[str, Any]:
    with _temporary_module_bindings(
        _docs_service,
        _chunk_text=_chunk_text,
        read_reference_url=_read_reference_url,
        read_reference_source=_read_reference_source,
    ):
        return _docs_service.ingest_reference_source(
            session_id,
            source_path=source_path,
            namespace=namespace,
            chunk_size=chunk_size,
            overlap=overlap,
            db_path=db_path,
        )

insert_model_state_snapshot

insert_model_state_snapshot(*, session_id: str, checkpoint_id: int | None = None, model_path: str | None = None, feature_tree_json: str | None = None, mass_properties_json: str | None = None, screenshot_path: str | None = None, state_fingerprint: str | None = None, db_path: Path | None = None) -> int

Insert model snapshot row and return snapshot ID for rollback tracking.

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
model_path	str | None	The model path value. Defaults to None.	None
feature_tree_json	str | None	The feature tree json value. Defaults to None.	None
mass_properties_json	str | None	The mass properties json value. Defaults to None.	None
screenshot_path	str | None	The screenshot path value. Defaults to None.	None
state_fingerprint	str | None	The state fingerprint 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_model_state_snapshot(
    *,
    session_id: str,
    checkpoint_id: int | None = None,
    model_path: str | None = None,
    feature_tree_json: str | None = None,
    mass_properties_json: str | None = None,
    screenshot_path: str | None = None,
    state_fingerprint: str | None = None,
    db_path: Path | None = None,
) -> int:
    """Insert model snapshot row and return snapshot ID for rollback tracking.

    Args:
        session_id (str): The session id value.
        checkpoint_id (int | None): The checkpoint id value. Defaults to None.
        model_path (str | None): The model path value. Defaults to None.
        feature_tree_json (str | None): The feature tree json value. Defaults to None.
        mass_properties_json (str | None): The mass properties json value. Defaults to None.
        screenshot_path (str | None): The screenshot path value. Defaults to None.
        state_fingerprint (str | None): The state fingerprint 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)
    with Session(engine) as session:
        row = ModelStateSnapshot(
            session_id=session_id,
            checkpoint_id=checkpoint_id,
            model_path=model_path,
            feature_tree_json=feature_tree_json,
            mass_properties_json=mass_properties_json,
            screenshot_path=screenshot_path,
            state_fingerprint=state_fingerprint,
            created_at=_utc_now_iso(),
        )
        session.add(row)
        session.commit()
        session.refresh(row)
        return int(row.id or 0)

inspect_family async

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

Source code in src/solidworks_mcp/ui/service.py

async def inspect_family(
    session_id: str,
    user_goal: str,
    *,
    model_name: str | None = None,
    db_path: Path | None = None,
) -> dict[str, Any]:
    async def _compat_run_structured_agent(*args: Any, **kwargs: Any):
        result = await _run_structured_agent(*args, **kwargs)
        if isinstance(result, RecoverableFailure):
            return _llm_service.RecoverableFailure(
                explanation=str(getattr(result, "explanation", "")),
                remediation_steps=list(getattr(result, "remediation_steps", []) or []),
                retry_focus=str(getattr(result, "retry_focus", "")) or "",
                should_retry=bool(getattr(result, "should_retry", False)),
            )
        if all(
            hasattr(result, name)
            for name in (
                "explanation",
                "remediation_steps",
                "retry_focus",
                "should_retry",
            )
        ):
            return _llm_service.RecoverableFailure(
                explanation=str(getattr(result, "explanation", "")),
                remediation_steps=list(getattr(result, "remediation_steps", []) or []),
                retry_focus=str(getattr(result, "retry_focus", "")),
                should_retry=bool(getattr(result, "should_retry", False)),
            )
        return result

    with _temporary_module_bindings(
        _llm_service, _run_structured_agent=_compat_run_structured_agent
    ):
        return await _llm_service.inspect_family(
            session_id,
            user_goal,
            model_name=model_name,
            db_path=db_path,
        )

is_url_reference

is_url_reference(source_path: str) -> bool

Return True when source_path is an http/https URL.

Parameters:

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

Returns:

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

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

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

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

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

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
    ]

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)

materialize_uploaded_model

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

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

Parameters:

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

Returns:

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

Raises:

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

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

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

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

    Returns:
        Path pointing at the decoded file on disk.

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

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

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

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

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

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

merge_metadata

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

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

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

Parameters:

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

Returns:

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

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

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

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

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

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

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

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

normalize_feature_targets

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

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

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

Parameters:

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

Returns:

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

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

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

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

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

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

normalize_workflow_mode

normalize_workflow_mode(workflow_mode: str | None) -> str

Normalise a raw workflow mode string to a known value.

Parameters:

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

Returns:

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

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

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

    Args:
        workflow_mode: Raw mode string from UI state.

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

open_target_model async

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

Source code in src/solidworks_mcp/ui/service.py

async def open_target_model(
    session_id: str,
    *,
    model_path: str | None = None,
    uploaded_files: list[dict[str, Any]] | None = None,
    feature_target_text: str = "",
    db_path: Path | None = None,
    api_origin: str = DEFAULT_API_ORIGIN,
) -> dict[str, Any]:
    with _temporary_module_bindings(
        _model_service,
        create_adapter=create_adapter,
        load_config=load_config,
    ):
        return await _model_service.open_target_model(
            session_id,
            model_path=model_path,
            uploaded_files=uploaded_files,
            feature_target_text=feature_target_text,
            db_path=db_path,
            api_origin=api_origin,
        )

parse_json_blob

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

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

Parameters:

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

Returns:

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

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

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

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

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

persist_ui_action

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

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

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

Parameters:

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

Returns:

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

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

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

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

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

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

    record_output = merged_metadata if output_metadata else output_payload

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

provider_from_model_name

provider_from_model_name(model_name: str) -> str

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

Parameters:

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

Returns:

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

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

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

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

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

provider_has_credentials

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

Check whether the required credentials exist for the given model.

Parameters:

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

Returns:

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

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

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

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

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

read_reference_source

read_reference_source(source_path: Path) -> str

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

PDF extraction requires the optional pypdf package.

Parameters:

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

Returns:

Type	Description
str	Extracted plain text content.

Raises:

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

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

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

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

    Args:
        source_path: Local file path to read.

    Returns:
        Extracted plain text content.

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

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

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

read_reference_url

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

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

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

Parameters:

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

Returns:

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

Raises:

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

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

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

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

    Args:
        source_url: http/https URL to fetch.

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

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

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

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

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

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

    return decoded.strip(), label

reconcile_manual_edits

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

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

Parameters:

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

Returns:

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

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

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

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

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

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

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

refresh_preview async

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

Source code in src/solidworks_mcp/ui/service.py

async def refresh_preview(
    session_id: str,
    *,
    orientation: str = DEFAULT_PREVIEW_ORIENTATION,
    db_path: Path | None = None,
    preview_dir: Path | None = None,
    api_origin: str = DEFAULT_API_ORIGIN,
) -> dict[str, Any]:
    with _temporary_module_bindings(
        _preview_service,
        create_adapter=create_adapter,
        load_config=load_config,
    ):
        return await _preview_service.refresh_preview(
            session_id,
            orientation=orientation,
            db_path=db_path,
            preview_dir=preview_dir,
            api_origin=api_origin,
        )

request_clarifications async

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

Source code in src/solidworks_mcp/ui/service.py

async def request_clarifications(
    session_id: str,
    user_goal: str,
    *,
    user_answer: str = "",
    model_name: str | None = None,
    db_path: Path | None = None,
) -> dict[str, Any]:
    async def _compat_run_structured_agent(*args: Any, **kwargs: Any):
        result = await _run_structured_agent(*args, **kwargs)
        if isinstance(result, RecoverableFailure):
            return _llm_service.RecoverableFailure(
                explanation=str(getattr(result, "explanation", "")),
                remediation_steps=list(getattr(result, "remediation_steps", []) or []),
                retry_focus=str(getattr(result, "retry_focus", "")) or "",
                should_retry=bool(getattr(result, "should_retry", False)),
            )
        if all(
            hasattr(result, name)
            for name in (
                "explanation",
                "remediation_steps",
                "retry_focus",
                "should_retry",
            )
        ):
            return _llm_service.RecoverableFailure(
                explanation=str(getattr(result, "explanation", "")),
                remediation_steps=list(getattr(result, "remediation_steps", []) or []),
                retry_focus=str(getattr(result, "retry_focus", "")),
                should_retry=bool(getattr(result, "should_retry", False)),
            )
        return result

    with _temporary_module_bindings(
        _llm_service, _run_structured_agent=_compat_run_structured_agent
    ):
        return await _llm_service.request_clarifications(
            session_id,
            user_goal,
            user_answer=user_answer,
            model_name=model_name,
            db_path=db_path,
        )

run_go_orchestration async

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

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

Parameters:

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

Returns:

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

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

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

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

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

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

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

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

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

safe_context_name

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

Normalise a context name to a safe filesystem slug.

Parameters:

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

Returns:

Type	Description
str	Alphanumeric + hyphen slug, never empty.

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

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

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

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

sanitize_model_path_text

sanitize_model_path_text(value: Any) -> str

Strip surrounding quotes from a model path string.

Parameters:

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

Returns:

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

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

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

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

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

sanitize_preview_viewer_url

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

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

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

Parameters:

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

Returns:

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

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

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

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

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

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

sanitize_ui_text

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

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

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

Parameters:

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

Returns:

Type	Description
str	Cleaned string, or fallback.

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

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

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

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

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

save_session_context

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

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

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

Parameters:

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

Returns:

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

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

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

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

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

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

select_workflow_mode

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

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

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

Parameters:

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

Returns:

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

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

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

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

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

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

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

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

trace_json

trace_json(value: Any) -> str

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

Parameters:

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

Returns:

Type	Description
str	Indented JSON string.

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

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

    Args:
        value: Any JSON-serialisable value.

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

trace_session_row

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

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

Parameters:

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

Returns:

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

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

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

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

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

trace_tool_records

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

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

Parameters:

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

Returns:

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

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

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

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

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

update_session_notes

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

Persist free-form engineering notes in session metadata.

Parameters:

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

Returns:

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

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

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

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

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

update_ui_preferences

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

Persist manufacturing assumptions and LLM provider preferences.

Parameters:

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

Returns:

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

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

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

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

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

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

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