import json import logging import re from typing import Any, Awaitable, Callable, Literal import dirtyjson # type: ignore[import-untyped] from llmai.shared import AssistantToolCall, Tool # type: ignore[import-not-found] from constants.presentation import MAX_NUMBER_OF_SLIDES, MAX_OUTLINE_CONTENT_WORDS from services.chat.schemas import ( AddElementInput, AddNewSlideInput, AddNewSlideLayoutInput, AddOutlineInput, AddSlideComponentInput, DeleteSlideComponentInput, DeleteSlideElementInput, DeleteSlideInput, DeleteOutlineInput, GenerateAssetsInput, GetAvailableBlocksInput, GetContentSchemaFromLayoutIdInput, GetSlideAtIndexInput, NoArgsInput, ReadSourceDocumentsInput, SaveSlideInput, SearchSlidesInput, SetPresentationThemeInput, UpdateComponentInput, UpdateSlideInput, UpdateSlideComponentInput, UpdateOutlineInput, UpdateSlideElementInput, ) from services.chat.presentation_context_store import PresentationContextStore LOGGER = logging.getLogger(__name__) ToolHandler = Callable[[dict[str, Any]], Awaitable[dict[str, Any]]] ChatToolMode = Literal["presentation", "outline"] CHART_INSERT_TOOL_FIELDS = { "addElement": "element", "addComponent": "component", "createComponent": "component", "updateComponent": "component", } TABLE_INSERT_TOOL_FIELDS = CHART_INSERT_TOOL_FIELDS IMAGE_INSERT_TOOL_FIELDS = CHART_INSERT_TOOL_FIELDS BLOCK_PRIORITIZED_INSERT_TYPES = {"chart", "table"} JSON_OBJECT_STRING_FIELDS = { "addNewSlideLayout": ("content",), "saveSlide": ("content",), "updateSlide": ("content",), "addElement": ("element",), "updateElement": ("element",), "addComponent": ("component",), "createComponent": ("component",), "updateComponent": ("component",), } MAX_TOOL_REPAIR_RETRIES = 1 def _normalize_tool_argument_value(value: Any) -> Any: if isinstance(value, str) and value.strip().lower() == "null": return None if isinstance(value, list): return [_normalize_tool_argument_value(item) for item in value] if isinstance(value, dict): return { key: _normalize_tool_argument_value(item) for key, item in value.items() } return value class ChatTools: def __init__( self, memory: PresentationContextStore, mode: ChatToolMode = "presentation", ): self._memory = memory self._mode = mode self._turn_user_message = "" self._generated_assets: list[dict[str, Any]] = [] self._tool_handlers: dict[str, ToolHandler] = { "addOutline": self._add_outline, "updateOutline": self._update_outline, "deleteOutline": self._delete_outline, "addNewSlide": self._add_new_slide, "addNewSlideLayout": self._add_new_slide_layout, "getTemplateSummary": self._get_template_summary, "readSourceDocuments": self._read_source_documents, "searchSlide": self._search_slides, "getSlideAtIndex": self._get_slide_at_index, "getAvailableLayouts": self._get_available_layouts, "getAvailableBlocks": self._get_available_blocks, "getContentSchemaFromLayoutId": self._get_content_schema_from_layout_id, "generateAssets": self._generate_assets, "saveSlide": self._save_slide, "updateSlide": self._update_slide, "deleteSlide": self._delete_slide, "addElement": self._add_element, "updateElement": self._update_slide_element, "deleteElement": self._delete_slide_element, "addComponent": self._add_slide_component, "createComponent": self._add_slide_component, "updateComponent": self._update_component, "deleteComponent": self._delete_slide_component, "getPresentationTheme": self._get_presentation_theme_catalog, "setPresentationTheme": self._set_presentation_theme, } def set_turn_context(self, user_message: str) -> None: self._turn_user_message = user_message or "" self._generated_assets = [] def get_tool_definitions(self) -> list[Tool]: return [ Tool( name="addOutline", description=( "Insert a new markdown outline item into the outline draft. " "This edits presentation.outlines only and does not require a layout. " f"Do not exceed {MAX_NUMBER_OF_SLIDES} outline slides or " f"{MAX_OUTLINE_CONTENT_WORDS} words in this outline item." ), schema=AddOutlineInput, strict=False, ), Tool( name="updateOutline", description=( "Replace the markdown content of one outline item by zero-based index. " "This edits presentation.outlines only and does not require a layout. " f"Keep this outline item within {MAX_OUTLINE_CONTENT_WORDS} words." ), schema=UpdateOutlineInput, strict=False, ), Tool( name="deleteOutline", description=( "Delete one outline item by zero-based index. This edits " "presentation.outlines only and does not require a layout." ), schema=DeleteOutlineInput, strict=False, ), Tool( name="addNewSlide", description=( "Add a blank slide to the current presentation at a zero-based index " "or append when index is null." ), schema=AddNewSlideInput, strict=False, ), Tool( name="addNewSlideLayout", description=( "Add a new slide from an available layout. Use getAvailableLayouts " "first, then pass content as a JSON-serialized object matching the layout." ), schema=AddNewSlideLayoutInput, strict=False, ), Tool( name="getAvailableLayouts", description="List available slide layout ids, names, and summaries.", schema=NoArgsInput, strict=False, ), Tool( name="getAvailableBlocks", description=( "Search reusable template component blocks without fetching a whole layout. " "Use this before addComponent/createComponent when the user asks to add " "a styled block such as a title/header/subtitle, table, chart, card, " "callout, metric, image panel, or repeated content item. For title " "blocks use elementType=text with title/header/heading query terms. " "Set includeFullContent=true only when exact component JSON is needed." ), schema=GetAvailableBlocksInput, strict=False, ), Tool( name="getContentSchemaFromLayoutId", description=( "Return the exact JSON content schema for one layout id. " "Use this before addNewSlideLayout or updateSlide when composing " "a full slide content payload." ), schema=GetContentSchemaFromLayoutIdInput, strict=False, ), Tool( name="getTemplateSummary", description=( "Read a compact summary of the current presentation template, " "layouts, current slides, and theme. Use before choosing where/how to edit." ), schema=NoArgsInput, strict=False, ), Tool( name="readSourceDocuments", description=( "Read parsed text from the source document(s) uploaded for this " "presentation, including PDFs. Use when the user refers to an " "uploaded/source PDF, document, file, or the document used to " "generate the deck. Use before summarizing, quoting, extracting " "data, or creating slide content from uploaded documents." ), schema=ReadSourceDocumentsInput, strict=False, ), Tool( name="searchSlide", description=( "Search current slides for text/topics and return slide indices and snippets." ), schema=SearchSlidesInput, strict=False, ), Tool( name="getSlideAtIndex", description=( "Live SQL: one slide by index—authoritative for exact current content. " "Set includeFullContent=true when you need full JSON (before saveSlide or precise edits). " "If user says slide N, use zero-based index N-1." ), schema=GetSlideAtIndexInput, strict=False, ), Tool( name="saveSlide", description=( "Save full slide content for a layout. Use for complete slide payloads; " "visible element/component edits should use element/component tools." ), schema=SaveSlideInput, strict=False, ), Tool( name="updateSlide", description="Replace an existing slide's layout/content by zero-based index.", schema=UpdateSlideInput, strict=False, ), Tool( name="deleteSlide", description="Delete an existing slide by zero-based index and reindex the rest.", schema=DeleteSlideInput, strict=False, ), Tool( name="addElement", description=( "Add one rendered UI element to a slide, either inside a componentId " "or as a new free component when componentId is null. Do not use this " "for new table/chart requests when a reusable block exists; use " "getAvailableBlocks and addComponent/createComponent instead. Chart " "elements must include numeric data as categories plus series.values, " "or legacy data rows with label/value. Image elements must include " "data set to a URL returned by generateAssets." ), schema=AddElementInput, strict=False, ), Tool( name="updateElement", description=( "Update visible element content or geometry using an elementPath returned " "by getSlideAtIndex. Supports text, lists, table, chart, image data, " "position, size, and toolbar-style font, color, fill, stroke, " "alignment, opacity, and element property patches. For " "charts, use the chart field for chartType, categories, " "series.values, colors, axes, dataLabels placement, and legend." ), schema=UpdateSlideElementInput, strict=False, ), Tool( name="deleteElement", description="Delete one rendered UI element by elementPath.", schema=DeleteSlideElementInput, strict=False, ), Tool( name="addComponent", description=( "Add an existing/new rendered UI component block to a slide. Component " "JSON must include id, description, position, size, and elements. Chart " "elements must include numeric data as categories plus series.values, or " "legacy data rows with label/value. Image elements must include data " "set to a URL returned by generateAssets. For styled title/header, " "table/chart, card, metric, or callout additions, adapt a component " "returned by getAvailableBlocks and pass sourceBlockId." ), schema=AddSlideComponentInput, strict=False, ), Tool( name="createComponent", description=( "Create a grouped rendered UI component from provided component JSON " "and add it to a slide. Chart elements must include numeric data as " "categories plus series.values, or legacy data rows with label/value. " "Image elements must include data set to a URL returned by generateAssets. " "For styled title/header, table/chart, card, metric, or callout additions " "adapt a component returned by getAvailableBlocks and pass sourceBlockId." ), schema=AddSlideComponentInput, strict=False, ), Tool( name="updateComponent", description=( "Move, resize, replace, duplicate, reorder, group, or ungroup rendered " "UI components by componentId." ), schema=UpdateComponentInput, strict=False, ), Tool( name="deleteComponent", description=( "Remove one whole component (a block such as a numbered point, card, " "or callout) from a rendered slide by componentId." ), schema=DeleteSlideComponentInput, strict=False, ), Tool( name="getPresentationTheme", description="Read the current presentation theme and available themes.", schema=NoArgsInput, strict=False, ), Tool( name="setPresentationTheme", description=( "Change the deck theme by theme name/id/query or customTheme payload." ), schema=SetPresentationThemeInput, strict=False, ), Tool( name="generateAssets", description="Generate one or more image/icon assets for slide edits.", schema=GenerateAssetsInput, strict=False, ), ] async def execute_tool_call(self, tool_call: AssistantToolCall) -> dict[str, Any]: handler = self._tool_handlers.get(tool_call.name) if not handler: return { "ok": False, "tool": tool_call.name, "error": f"Unsupported tool: {tool_call.name}", "recovery": { "retryable": False, "message": "Use one of the available chat tools.", "guidance": ["Choose a tool from the tool definitions."], }, } parsed_args: dict[str, Any] | None = None repair_notes: list[str] = [] try: parsed_args = self._parse_args(tool_call.arguments) parsed_args, repair_notes = self._repair_tool_args( tool_call.name, parsed_args, ) LOGGER.info("Executing chat tool %s", tool_call.name) try: result = await handler(parsed_args) except Exception as first_exc: retried = False for _attempt in range(MAX_TOOL_REPAIR_RETRIES): retry_args, retry_notes = self._repair_tool_args( tool_call.name, parsed_args, error=str(first_exc), ) if not retry_notes or self._args_equivalent(parsed_args, retry_args): break repair_notes.extend(retry_notes) parsed_args = retry_args LOGGER.info( "Retrying chat tool %s after argument repair", tool_call.name, ) result = await handler(parsed_args) retried = True break if not retried: raise first_exc if tool_call.name == "generateAssets": self._remember_generated_assets(result) response = {"ok": True, "tool": tool_call.name, "result": result} if repair_notes: response["repair"] = { "applied": True, "notes": repair_notes, } return response except Exception as exc: LOGGER.exception("Chat tool failed: %s", tool_call.name) return { "ok": False, "tool": tool_call.name, "error": str(exc), "repair": { "attempted": bool(repair_notes), "notes": repair_notes, }, "recovery": self._build_tool_recovery( tool_name=tool_call.name, args=parsed_args, error=str(exc), ), } async def _get_presentation_outline(self, _: dict[str, Any]) -> dict[str, Any]: outline = await self._memory.get("presentation_outline") if not isinstance(outline, dict): return { "found": False, "message": "Presentation outline is not available in memory yet.", "sections": [], } slides = outline.get("slides") if not isinstance(slides, list) or not slides: return { "found": False, "message": "Presentation outline exists but has no slides.", "sections": [], } sections: list[dict[str, Any]] = [] for position, slide in enumerate(slides): index = position content = "" if isinstance(slide, dict): raw_index = slide.get("index") if isinstance(raw_index, int): index = raw_index raw_content = slide.get("content") if isinstance(raw_content, str): content = raw_content elif raw_content is not None: try: content = json.dumps(raw_content, ensure_ascii=False) except Exception: content = str(raw_content) elif isinstance(slide, str): content = slide title = self._extract_title(content) or f"Slide {index + 1}" sections.append( { "index": index, "slide_number": index + 1, "title": title, } ) return { "found": True, "slide_count": len(sections), "sections": sections, "source": outline.get("source", "memory"), } async def _search_slides(self, args: dict[str, Any]) -> dict[str, Any]: payload = SearchSlidesInput(**args) results = await self._memory.search(payload.query, payload.limit) return { "query": payload.query, "count": len(results), "results": results, } async def _get_slide_at_index(self, args: dict[str, Any]) -> dict[str, Any]: normalized_args = dict(args) normalized_args.setdefault("includeFullContent", False) payload = GetSlideAtIndexInput(**normalized_args) slide = await self._memory.get_slide_at_index( payload.index, include_full_content=payload.include_full_content, ) if not slide and payload.index > 0: # Users often refer to slides as 1-based; allow a safe fallback. fallback_index = payload.index - 1 fallback_slide = await self._memory.get_slide_at_index( fallback_index, include_full_content=payload.include_full_content, ) if fallback_slide: return { "found": True, "slide": fallback_slide, "requested_index": payload.index, "resolved_index": fallback_index, "note": ( "No slide found at requested index; returned one-based fallback " f"at index {fallback_index}." ), } if not slide: return { "found": False, "message": f"No slide found at index {payload.index}.", } return { "found": True, "slide": slide, } async def _get_outline_draft(self, _: dict[str, Any]) -> dict[str, Any]: return await self._memory.get_outline_draft() async def _add_outline(self, args: dict[str, Any]) -> dict[str, Any]: payload = AddOutlineInput(**args) return await self._memory.add_outline( content=payload.content, index=payload.index, ) async def _update_outline(self, args: dict[str, Any]) -> dict[str, Any]: payload = UpdateOutlineInput(**args) return await self._memory.update_outline( index=payload.index, content=payload.content, ) async def _delete_outline(self, args: dict[str, Any]) -> dict[str, Any]: payload = DeleteOutlineInput(**args) return await self._memory.delete_outline(index=payload.index) async def _add_new_slide(self, args: dict[str, Any]) -> dict[str, Any]: payload = AddNewSlideInput(**args) return await self._memory.add_blank_slide(index=payload.index) async def _add_new_slide_layout(self, args: dict[str, Any]) -> dict[str, Any]: payload_args = json.loads(json.dumps(dict(args), ensure_ascii=False)) raw_content = payload_args.get("content") if isinstance(raw_content, dict): payload_args["content"] = json.dumps(raw_content, ensure_ascii=False) payload = AddNewSlideLayoutInput(**payload_args) return await self._save_slide( { "content": payload.content, "layoutId": payload.layout_id, "index": payload.index, "replaceOldSlideAtIndex": False, } ) async def _update_slide(self, args: dict[str, Any]) -> dict[str, Any]: payload_args = json.loads(json.dumps(dict(args), ensure_ascii=False)) raw_content = payload_args.get("content") if isinstance(raw_content, dict): payload_args["content"] = json.dumps(raw_content, ensure_ascii=False) payload = UpdateSlideInput(**payload_args) return await self._save_slide( { "content": payload.content, "layoutId": payload.layout_id, "index": payload.index, "replaceOldSlideAtIndex": True, } ) async def _get_available_layouts(self, _: dict[str, Any]) -> dict[str, Any]: layouts = await self._memory.get_available_layouts() return { "count": len(layouts), "layouts": layouts, } async def _get_available_blocks(self, args: dict[str, Any]) -> dict[str, Any]: payload = GetAvailableBlocksInput(**args) max_results = payload.max_results if payload.max_results is not None else 20 return await self._memory.get_available_blocks( query=payload.query, layout_id=payload.layout_id, element_type=payload.element_type, block_id=payload.block_id, include_full_content=bool(payload.include_full_content), max_results=max_results, ) async def _get_template_summary(self, _: dict[str, Any]) -> dict[str, Any]: outline = await self._get_presentation_outline({}) layouts = await self._get_available_layouts({}) theme = await self._get_presentation_theme_catalog({}) return { "outline": outline, "available_layouts": layouts, "theme": theme, "message": "Template summary fetched successfully.", } async def _read_source_documents(self, args: dict[str, Any]) -> dict[str, Any]: payload = ReadSourceDocumentsInput(**args) return await self._memory.read_source_documents( query=payload.query, max_chars=payload.max_chars, ) async def _get_presentation_theme_catalog( self, _: dict[str, Any] ) -> dict[str, Any]: return await self._memory.get_presentation_theme_catalog() async def _get_content_schema_from_layout_id( self, args: dict[str, Any] ) -> dict[str, Any]: payload = GetContentSchemaFromLayoutIdInput(**args) schema = await self._memory.get_content_schema_from_layout_id(payload.layout_id) if schema is None: return { "found": False, "layout_id": payload.layout_id, "message": "Layout schema not found for the provided layout id.", } return { "found": True, "layout_id": payload.layout_id, "content_schema": schema, } async def _generate_assets(self, args: dict[str, Any]) -> dict[str, Any]: payload = GenerateAssetsInput(**args) generated_assets: list[dict[str, Any]] = [] for index, asset in enumerate(payload.assets): if asset.kind == "image": url = await self._memory.generate_image(asset.prompt) else: url = await self._memory.generate_icon(asset.prompt) generated_assets.append( { "index": index, "kind": asset.kind, "prompt": asset.prompt, "url": url, } ) return { "count": len(generated_assets), "assets": generated_assets, "message": f"Generated {len(generated_assets)} asset(s).", } async def _save_slide(self, args: dict[str, Any]) -> dict[str, Any]: payload_args = json.loads(json.dumps(dict(args), ensure_ascii=False)) raw_content = payload_args.get("content") if isinstance(raw_content, dict): payload_args["content"] = json.dumps(raw_content, ensure_ascii=False) payload = SaveSlideInput(**payload_args) try: content_parsed: Any = dirtyjson.loads(payload.content) except Exception: content_parsed = json.loads(payload.content) if not isinstance(content_parsed, dict): raise ValueError("'content' must be a JSON object.") content_payload = json.loads(json.dumps(content_parsed, ensure_ascii=False)) return await self._memory.save_slide( content=content_payload, layout_id=payload.layout_id, index=payload.index, replace_old_slide_at_index=payload.replace_old_slide_at_index, ) async def _delete_slide(self, args: dict[str, Any]) -> dict[str, Any]: payload = DeleteSlideInput(**args) return await self._memory.delete_slide(index=payload.index) async def _get_slide_elements(self, args: dict[str, Any]) -> dict[str, Any]: payload = GetSlideAtIndexInput( index=int(args.get("index") or 0), includeFullContent=bool(args.get("includeFullJson")), ) return await self._memory.get_slide_ui_elements( index=payload.index, include_full_json=payload.include_full_content, ) async def _add_element(self, args: dict[str, Any]) -> dict[str, Any]: payload = AddElementInput(**args) try: parsed: Any = dirtyjson.loads(payload.element) except Exception: parsed = json.loads(payload.element) element = json.loads(json.dumps(parsed, ensure_ascii=False)) if not isinstance(element, dict): raise ValueError("'element' must be a JSON object.") await self._require_reusable_block_first( tree=element, source_block_id=None, primitive_tool="addElement", ) return await self._memory.add_slide_ui_element( index=payload.index, element=element, component_id=payload.component_id, insert_index=payload.insert_index, ) async def _update_slide_element(self, args: dict[str, Any]) -> dict[str, Any]: payload = UpdateSlideElementInput(**args) element_patch: dict[str, Any] | None = None if payload.element is not None: try: parsed: Any = dirtyjson.loads(payload.element) except Exception: parsed = json.loads(payload.element) element_patch = json.loads(json.dumps(parsed, ensure_ascii=False)) if not isinstance(element_patch, dict): raise ValueError("'element' must be a JSON object.") style_patch = self._element_style_patch_from_update_payload(payload) if style_patch: element_patch = self._merge_dict_patch(element_patch or {}, style_patch) return await self._memory.update_slide_ui_element( index=payload.index, element_path=payload.element_path, text=payload.text, items=payload.items, table_cell=( payload.table_cell.model_dump(by_alias=False) if payload.table_cell is not None else None ), table=( payload.table.model_dump() if payload.table is not None else None ), chart=( payload.chart.model_dump(exclude_none=True) if payload.chart is not None else None ), element_patch=element_patch, position=( payload.position.model_dump() if payload.position is not None else None ), size=payload.size.model_dump() if payload.size is not None else None, ) async def _update_slide_component(self, args: dict[str, Any]) -> dict[str, Any]: payload = UpdateSlideComponentInput(**args) return await self._memory.update_slide_ui_component( index=payload.index, component_id=payload.component_id, position=( payload.position.model_dump() if payload.position is not None else None ), size=payload.size.model_dump() if payload.size is not None else None, ) async def _update_component(self, args: dict[str, Any]) -> dict[str, Any]: payload = UpdateComponentInput(**args) replacement_component: dict[str, Any] | None = None if payload.component is not None: try: parsed: Any = dirtyjson.loads(payload.component) except Exception: parsed = json.loads(payload.component) replacement_component = json.loads(json.dumps(parsed, ensure_ascii=False)) if not isinstance(replacement_component, dict): raise ValueError("'component' must be a JSON object.") return await self._memory.update_slide_ui_component( index=payload.index, component_id=payload.component_id, action=payload.action or "update", component_ids=payload.component_ids, position=( payload.position.model_dump() if payload.position is not None else None ), size=payload.size.model_dump() if payload.size is not None else None, replacement_component=replacement_component, ) async def _delete_slide_component(self, args: dict[str, Any]) -> dict[str, Any]: payload = DeleteSlideComponentInput(**args) return await self._memory.delete_slide_ui_component( index=payload.index, component_id=payload.component_id, ) async def _delete_slide_element(self, args: dict[str, Any]) -> dict[str, Any]: payload = DeleteSlideElementInput(**args) return await self._memory.delete_slide_ui_element( index=payload.index, element_path=payload.element_path, ) async def _add_slide_component(self, args: dict[str, Any]) -> dict[str, Any]: payload = AddSlideComponentInput(**args) try: component_parsed: Any = dirtyjson.loads(payload.component) except Exception: component_parsed = json.loads(payload.component) if not isinstance(component_parsed, dict): raise ValueError("'component' must be a JSON object.") component_payload = json.loads(json.dumps(component_parsed, ensure_ascii=False)) await self._require_reusable_block_first( tree=component_payload, source_block_id=payload.source_block_id, primitive_tool="addComponent", ) return await self._memory.add_slide_ui_component( index=payload.index, component=component_payload, insert_index=payload.insert_index, ) async def _set_presentation_theme(self, args: dict[str, Any]) -> dict[str, Any]: payload = SetPresentationThemeInput(**args) return await self._memory.set_presentation_theme( theme_query=payload.theme, custom_theme=( payload.custom_theme.model_dump(exclude_none=True) if payload.custom_theme is not None else None ), save_custom_theme=bool(payload.save_custom_theme), ) @staticmethod def _parse_args(arguments: str | None) -> dict[str, Any]: if not arguments: return {} parsed = ChatTools._loads_jsonish(arguments) normalized = _normalize_tool_argument_value( json.loads(json.dumps(parsed, ensure_ascii=False)) ) if isinstance(normalized, dict): return normalized raise ValueError("Tool arguments must be a JSON object.") async def _require_reusable_block_first( self, *, tree: dict[str, Any], source_block_id: str | None, primitive_tool: str, ) -> None: requested_types = self._block_prioritized_element_types(tree) if not requested_types: return if source_block_id: block_result = await self._memory.get_available_blocks( block_id=source_block_id, include_full_content=False, max_results=1, ) blocks = ( block_result.get("blocks") if isinstance(block_result, dict) else None ) block = blocks[0] if isinstance(blocks, list) and blocks else None if not isinstance(block, dict): raise ValueError( "sourceBlockId was provided but no matching reusable block was found. " "Call getAvailableBlocks again and use a returned block_id." ) block_types = { str(item).lower() for item in block.get("element_types", []) if item is not None } if requested_types.isdisjoint(block_types): raise ValueError( "sourceBlockId does not match the table/chart type being inserted. " "Use a block_id whose element_types include the requested type." ) return reusable = await self._first_available_reusable_block(requested_types) if reusable is None: return element_type, block = reusable block_id = str(block.get("block_id") or "") component_id = str(block.get("component_id") or "") layout_id = str(block.get("layout_id") or "") raise ValueError( f"Reusable block available for {element_type} insertion " f"(block_id='{block_id}', component_id='{component_id}', layout_id='{layout_id}'). " f"Do not use {primitive_tool} to create this as a primitive. " "Call getAvailableBlocks with that blockId and includeFullContent=true, " "adapt the returned component JSON with the requested content, then call " f"addComponent/createComponent with sourceBlockId='{block_id}'." ) async def _first_available_reusable_block( self, requested_types: set[str], ) -> tuple[str, dict[str, Any]] | None: for element_type in sorted(requested_types): block_result = await self._memory.get_available_blocks( element_type=element_type, include_full_content=False, max_results=1, ) blocks = ( block_result.get("blocks") if isinstance(block_result, dict) else None ) block = blocks[0] if isinstance(blocks, list) and blocks else None if isinstance(block, dict): return element_type, block return None @staticmethod def _block_prioritized_element_types(tree: Any) -> set[str]: found: set[str] = set() def visit(value: Any) -> None: if isinstance(value, dict): element_type = value.get("type") if isinstance(element_type, str): normalized = element_type.strip().lower() if normalized in BLOCK_PRIORITIZED_INSERT_TYPES: found.add(normalized) for nested in value.values(): visit(nested) elif isinstance(value, list): for nested in value: visit(nested) visit(tree) return found @staticmethod def _element_style_patch_from_update_payload( payload: UpdateSlideElementInput, ) -> dict[str, Any]: patch: dict[str, Any] = {} if payload.font is not None: font = payload.font.model_dump(exclude_none=True) if font: patch["font"] = font if payload.alignment is not None: alignment = payload.alignment.model_dump(exclude_none=True) if alignment: patch["alignment"] = alignment if payload.fill is not None: fill = payload.fill.model_dump(exclude_none=True) if fill: patch["fill"] = fill if payload.stroke is not None: stroke = payload.stroke.model_dump(exclude_none=True) if stroke: patch["stroke"] = stroke if payload.color is not None: patch["color"] = payload.color if payload.opacity is not None: patch["opacity"] = payload.opacity return patch @staticmethod def _merge_dict_patch( target: dict[str, Any], patch: dict[str, Any], ) -> dict[str, Any]: merged = json.loads(json.dumps(target, ensure_ascii=False)) for key, value in patch.items(): if isinstance(value, dict) and isinstance(merged.get(key), dict): merged[key] = ChatTools._merge_dict_patch(merged[key], value) else: merged[key] = json.loads(json.dumps(value, ensure_ascii=False)) return merged def _repair_tool_args( self, tool_name: str, args: dict[str, Any], *, error: str | None = None, ) -> tuple[dict[str, Any], list[str]]: repaired = dict(args) notes: list[str] = [] self._repair_json_object_string_fields(tool_name, repaired, notes) repaired = self._repair_chart_insert_args( tool_name, repaired, notes, error=error, ) repaired = self._repair_table_insert_args( tool_name, repaired, notes, error=error, ) repaired = self._repair_image_insert_args( tool_name, repaired, notes, error=error, ) return repaired, notes @staticmethod def _repair_json_object_string_fields( tool_name: str, args: dict[str, Any], notes: list[str], ) -> None: for field_name in JSON_OBJECT_STRING_FIELDS.get(tool_name, ()): if field_name not in args: continue value = args.get(field_name) if value is None: continue if isinstance(value, dict): args[field_name] = json.dumps(value, ensure_ascii=False) notes.append( f"Converted {field_name} from object to JSON string." ) continue if not isinstance(value, str): continue parsed = ChatTools._loads_jsonish_object(value) if parsed is None: continue canonical = json.dumps(parsed, ensure_ascii=False) if canonical != value: args[field_name] = canonical notes.append(f"Repaired JSON string field {field_name}.") def _repair_chart_insert_args( self, tool_name: str, args: dict[str, Any], notes: list[str], *, error: str | None = None, ) -> dict[str, Any]: payload_field = CHART_INSERT_TOOL_FIELDS.get(tool_name) if not payload_field or payload_field not in args: return args chart_rows = self._extract_chart_rows_from_user_message( self._turn_user_message, ) if not chart_rows: return args payload = self._parse_json_object_field(args.get(payload_field)) if payload is None: return args title = self._infer_chart_title_from_user_message(self._turn_user_message) if not self._inject_missing_chart_data(payload, chart_rows, title): return args repaired = dict(args) repaired[payload_field] = json.dumps(payload, ensure_ascii=False) notes.append( "Filled missing chart categories and series.values from the latest user message." ) return repaired def _repair_table_insert_args( self, tool_name: str, args: dict[str, Any], notes: list[str], *, error: str | None = None, ) -> dict[str, Any]: payload_field = TABLE_INSERT_TOOL_FIELDS.get(tool_name) if not payload_field or payload_field not in args: return args table_data = self._extract_table_from_user_message(self._turn_user_message) if table_data is None: return args payload = self._parse_json_object_field(args.get(payload_field)) if payload is None: return args if not self._inject_missing_table_data(payload, table_data): return args repaired = dict(args) repaired[payload_field] = json.dumps(payload, ensure_ascii=False) notes.append( "Filled missing table headers/columns and rows from the latest user message." ) return repaired def _repair_image_insert_args( self, tool_name: str, args: dict[str, Any], notes: list[str], *, error: str | None = None, ) -> dict[str, Any]: payload_field = IMAGE_INSERT_TOOL_FIELDS.get(tool_name) if not payload_field or payload_field not in args: return args payload = self._parse_json_object_field(args.get(payload_field)) if payload is None: return args if not self._inject_missing_image_data(payload): return args repaired = dict(args) repaired[payload_field] = json.dumps(payload, ensure_ascii=False) notes.append( "Filled missing image data from the generated asset URL in this turn." ) return repaired @staticmethod def _parse_json_object_field(value: Any) -> dict[str, Any] | None: if isinstance(value, dict): return json.loads(json.dumps(value, ensure_ascii=False)) if not isinstance(value, str) or not value.strip(): return None parsed = ChatTools._loads_jsonish_object(value) if not isinstance(parsed, dict): return None return json.loads(json.dumps(parsed, ensure_ascii=False)) def _remember_generated_assets(self, result: dict[str, Any]) -> None: assets = result.get("assets") if isinstance(result, dict) else None if not isinstance(assets, list): return for asset in assets: if not isinstance(asset, dict): continue url = asset.get("url") if not isinstance(url, str) or not url.strip(): continue self._generated_assets.append( { "kind": str(asset.get("kind") or "image"), "prompt": str(asset.get("prompt") or ""), "url": url.strip(), } ) @staticmethod def _loads_jsonish_object(value: str) -> dict[str, Any] | None: try: parsed = ChatTools._loads_jsonish(value) except Exception: return None return parsed if isinstance(parsed, dict) else None @staticmethod def _loads_jsonish(value: str) -> Any: last_exc: Exception | None = None for candidate in ChatTools._jsonish_candidates(value): try: return dirtyjson.loads(candidate) except Exception as exc: last_exc = exc try: return json.loads(candidate) except Exception as exc: last_exc = exc if last_exc: raise last_exc raise ValueError("JSON value is empty.") @staticmethod def _jsonish_candidates(value: str) -> list[str]: stripped = (value or "").strip() if not stripped: return [] candidates = [stripped] fence_match = re.search( r"```(?:json|javascript|js)?\s*(.*?)```", stripped, flags=re.IGNORECASE | re.DOTALL, ) if fence_match: candidates.append(fence_match.group(1).strip()) for opener, closer in (("{", "}"), ("[", "]")): start = stripped.find(opener) end = stripped.rfind(closer) if start != -1 and end > start: candidates.append(stripped[start : end + 1].strip()) unique: list[str] = [] seen: set[str] = set() for candidate in candidates: if not candidate or candidate in seen: continue seen.add(candidate) unique.append(candidate) return unique @classmethod def _inject_missing_chart_data( cls, node: Any, rows: list[dict[str, Any]], title: str, ) -> bool: from services.chat.slide_ui_helpers import _chart_element_has_explicit_data changed = False if isinstance(node, dict): if node.get("type") == "chart" and not _chart_element_has_explicit_data(node): categories = [row["label"] for row in rows] values = [row["value"] for row in rows] node.setdefault("chart_type", "bar") node.setdefault("title", title) node["categories"] = categories node["series"] = [{"name": title or "Series 1", "values": values}] node["data"] = [ {"label": row["label"], "value": row["value"]} for row in rows ] changed = True for value in node.values(): changed = cls._inject_missing_chart_data(value, rows, title) or changed elif isinstance(node, list): for value in node: changed = cls._inject_missing_chart_data(value, rows, title) or changed return changed @classmethod def _inject_missing_table_data( cls, node: Any, table_data: dict[str, Any], ) -> bool: from services.chat.slide_ui_helpers import _table_element_has_explicit_data changed = False if isinstance(node, dict): if node.get("type") == "table" and not _table_element_has_explicit_data(node): columns = list(table_data["columns"]) rows = [list(row) for row in table_data["rows"]] node["columns"] = columns node["rows"] = rows node.setdefault("min_columns", 1) node.setdefault("max_columns", max(len(columns), 1)) node.setdefault("min_rows", 1) node.setdefault("max_rows", max(len(rows), 1)) changed = True for value in node.values(): changed = cls._inject_missing_table_data(value, table_data) or changed elif isinstance(node, list): for value in node: changed = cls._inject_missing_table_data(value, table_data) or changed return changed def _inject_missing_image_data(self, node: Any) -> bool: from services.chat.slide_ui_helpers import _image_element_has_explicit_data changed = False if isinstance(node, dict): if node.get("type") == "image" and not _image_element_has_explicit_data(node): asset = self._latest_generated_asset_for_image( is_icon=node.get("is_icon") is True, ) if asset is not None: node["data"] = asset["url"] node.setdefault("is_icon", asset.get("kind") == "icon") prompt = asset.get("prompt") if isinstance(prompt, str) and prompt.strip(): node.setdefault("prompt", prompt.strip()) changed = True for value in node.values(): changed = self._inject_missing_image_data(value) or changed elif isinstance(node, list): for value in node: changed = self._inject_missing_image_data(value) or changed return changed def _latest_generated_asset_for_image( self, *, is_icon: bool, ) -> dict[str, Any] | None: preferred_kind = "icon" if is_icon else "image" for asset in reversed(self._generated_assets): if asset.get("kind") == preferred_kind and asset.get("url"): return asset for asset in reversed(self._generated_assets): if asset.get("url"): return asset return None @classmethod def _extract_chart_rows_from_user_message( cls, user_message: str, ) -> list[dict[str, Any]]: text = cls._strip_ui_context_prefix(user_message) if not text: return [] rows: list[dict[str, Any]] = [] seen_labels: set[str] = set() segments = re.split(r"[\n;,|]+|\s+\band\b\s+", text, flags=re.IGNORECASE) for segment in segments: segment = segment.strip(" \t\r\n.:-–—") if not segment: continue for match in re.finditer( r"(?P