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