fix: sanitize braces to avoid variable identification

agents/matmaster_agent/core_agents/comp_agents/recommend_summary_agent/tool_call_info_agent/prompt.py

@@ -1,6 +1,12 @@
+from agents.matmaster_agent.utils.sanitize_braces import with_sanitized_braces
+
+
+@with_sanitized_braces('user_prompt', 'agent_prompt')
 def gen_tool_call_info_instruction(
     user_prompt, agent_prompt, tool_doc, tool_schema, tool_args_recommend_prompt
 ):
+    user_prompt = user_prompt or ''
+    agent_prompt = agent_prompt or ''
     return f"""
 You are an AI agent that matches user requests to available tools. Your task is to analyze the user's query against the complete parameter schema.
 

agents/matmaster_agent/flow_agents/agent.py

@@ -147,6 +147,7 @@
     ReportUploadParams,
     upload_report_md_to_oss,
 )
+from agents.matmaster_agent.utils.sanitize_braces import sanitize_braces
 
 logger = logging.getLogger(__name__)
 logger.addFilter(PrefixFilter(MATMASTER_AGENT_NAME))
@@ -370,9 +371,9 @@ async def _run_expand_agent(
             yield expand_event
 
     async def _build_icl_prompt(self, ctx: InvocationContext):
-        UPDATE_USER_CONTENT = (
-            '\nUSER INPUT FOR THIS TASK:\n'
-            + ctx.session.state['expand']['update_user_content']
+        raw_content = ctx.session.state['expand']['update_user_content']
+        UPDATE_USER_CONTENT = '\nUSER INPUT FOR THIS TASK:\n' + sanitize_braces(
+            raw_content
         )
         icl_update_examples = select_update_examples(
             ctx.session.state['expand']['update_user_content'],

agents/matmaster_agent/flow_agents/analysis_agent/prompt.py

@@ -1,9 +1,14 @@
+from agents.matmaster_agent.utils.sanitize_braces import sanitize_braces
+
+
 def get_analysis_instruction(plan):
+    plan_str = plan if isinstance(plan, str) else str(plan)
+    plan_str = sanitize_braces(plan_str)
     return f"""
 分析 plan 变量的结果和之前的对话历史，来对本次计划的执行进行总结后展示给用户：
 
 <Plan Content>
-{plan}
+{plan_str}
 
 <Plan Structure>
 {{

agents/matmaster_agent/flow_agents/execution_agent/agent.py

@@ -45,6 +45,7 @@
     context_function_event,
     update_state_event,
 )
+from agents.matmaster_agent.utils.sanitize_braces import sanitize_braces
 
 logger = logging.getLogger(__name__)
 logger.addFilter(PrefixFilter(MATMASTER_AGENT_NAME))
@@ -211,8 +212,15 @@ async def _tool_result_validation(
         current_tool_description = ctx.session.state[PLAN]['steps'][index][
             STEP_DESCRIPTION
         ]
+        user_text = (
+            ctx.user_content.parts[0].text
+            if ctx.user_content and ctx.user_content.parts
+            else ''
+        )
+        user_text = sanitize_braces(user_text)
+        current_tool_description = sanitize_braces(current_tool_description or '')
         lines = (
-            f"用户原始请求: {ctx.user_content.parts[0].text}",
+            f"用户原始请求: {user_text}",
             f"当前步骤描述: {current_tool_description}",
             f"工具名称: {current_tool_name}",
             '请根据以上信息判断，工具的参数配置及对应的执行结果是否严格满足用户原始需求。',

agents/matmaster_agent/flow_agents/expand_agent/prompt.py

@@ -1,3 +1,5 @@
+from agents.matmaster_agent.utils.sanitize_braces import with_sanitized_braces
+
 # --- Context section headers (injected before EXPAND_INSTRUCTION when present) ---
 SECTION_SHORT_TERM_MEMORY = 'SHORT-TERM WORKING MEMORY'
 SECTION_SESSION_FILES = 'SESSION FILES'
@@ -13,6 +15,7 @@
 )
 
 
+@with_sanitized_braces('short_term_memory_block', 'session_file_summary')
 def build_expand_context(
     short_term_memory_block: str = '',
     session_file_summary: str = '',

agents/matmaster_agent/flow_agents/plan_make_agent/prompt.py

@@ -1,3 +1,6 @@
+from agents.matmaster_agent.utils.sanitize_braces import with_sanitized_braces
+
+
 def get_static_plan_system_block(available_tools_with_info: str) -> str:
     """
     Immutable content: persona, tool list (heaviest component), output format and constraints.
@@ -100,6 +103,7 @@ def get_static_plan_system_block(available_tools_with_info: str) -> str:
 """
 
 
+@with_sanitized_braces('thinking_context', 'session_file_summary', 'short_term_memory')
 def get_dynamic_plan_user_block(
     thinking_context: str = '',
     session_file_summary: str = '',

agents/matmaster_agent/flow_agents/report_agent/prompt.py

@@ -1,8 +1,11 @@
 from typing import Any, Mapping
 
+from agents.matmaster_agent.utils.sanitize_braces import sanitize_braces
+
 
 def get_report_instruction(plan: Mapping[str, Any]) -> str:
     """Return an instruction prompt to generate a self-contained markdown report."""
+    plan_str = sanitize_braces(str(plan))
 
     return f"""
 
@@ -115,7 +118,7 @@ def get_report_instruction(plan: Mapping[str, Any]) -> str:
 - [网页标题](URL) - 说明它支持了报告中的哪条事实/结论
 
 <Plan Content>
-{plan}
+{plan_str}
 
 （提示：你可以从对话/工具输出中抽取“用户输入文件”“执行时间”“执行概况”等信息；无法确定就用“-”。）
 

agents/matmaster_agent/flow_agents/thinking_agent/prompt.py

@@ -7,6 +7,8 @@
 
 from typing import TypedDict
 
+from agents.matmaster_agent.utils.sanitize_braces import with_sanitized_braces
+
 
 class ThinkingPromptBlocks(TypedDict):
     """Separated blocks for system (static) and user (dynamic) message assembly."""
@@ -92,6 +94,12 @@ def get_static_system_block(available_tools_with_info: str) -> str:
 """
 
 
+@with_sanitized_braces(
+    'session_file_summary',
+    'original_query',
+    'expanded_query',
+    'short_term_memory',
+)
 def get_dynamic_user_block(
     session_file_summary: str,
     original_query: str,
@@ -202,6 +210,13 @@ def get_static_revision_system_block(available_tools_with_info: str) -> str:
 """
 
 
+@with_sanitized_braces(
+    'session_file_summary',
+    'original_query',
+    'expanded_query',
+    'previous_reasoning',
+    'short_term_memory',
+)
 def get_dynamic_revision_user_block(
     session_file_summary: str,
     original_query: str,

agents/matmaster_agent/memory/prompt.py

@@ -1,10 +1,15 @@
+from agents.matmaster_agent.utils.sanitize_braces import (
+    with_sanitized_braces,
+)
+
 # Threshold (chars) above which we treat user context as "literature / long context".
 LONG_CONTEXT_THRESHOLD = 2500
 
 # Max insights for long context (literature / expert intuition).
 LONG_CONTEXT_MAX_INSIGHTS = 25
 
 
+@with_sanitized_braces('user_context', 'plan_intro')
 def get_memory_writer_instruction(
     user_context: str,
     plan_intro: str,

agents/matmaster_agent/utils/sanitize_braces.py

@@ -0,0 +1,63 @@
+"""
+Escape curly braces in dynamic text so {xx} is not interpreted as a variable placeholder.
+
+Used when embedding user/model content into templates that use {variable} substitution
+(e.g. ADK instruction inject_session_state). If the runtime is changed later, remove
+the decorator or calls and braces no longer need escaping.
+"""
+
+import functools
+import inspect
+
+
+def sanitize_braces(text: str) -> str:
+    """Escape `{` and `}` in text so they are not treated as variable placeholders.
+
+    Use for any dynamic content (user input, model output, plan text, etc.) before
+    inserting into a template that does {name} substitution. Empty/None returns as-is.
+    """
+    if not text:
+        return text
+    return text.replace('\\', '\\\\').replace('{', '\\{').replace('}', '\\}')
+
+
+def with_sanitized_braces(*param_names: str):
+    """Decorator: sanitize the listed string parameters before calling the function.
+
+    Use on functions that build instruction/template strings from dynamic args.
+    If you stop using a runtime that interprets {var}, remove this decorator and
+    no other logic changes are needed.
+    """
+    param_set = set(param_names)
+
+    def deco(f):
+        sig = inspect.signature(f)
+
+        @functools.wraps(f)
+        def wrapper(*args, **kwargs):
+            bound = sig.bind(*args, **kwargs)
+            bound.apply_defaults()
+            for name in param_set:
+                if name in bound.arguments and isinstance(bound.arguments[name], str):
+                    bound.arguments[name] = sanitize_braces(bound.arguments[name])
+            args_list = []
+            kwargs_dict = {}
+            for name in sig.parameters:
+                p = sig.parameters[name]
+                val = bound.arguments.get(name)
+                if p.kind == inspect.Parameter.VAR_POSITIONAL:
+                    args_list.extend(val)
+                elif p.kind == inspect.Parameter.VAR_KEYWORD:
+                    kwargs_dict.update(val)
+                elif p.kind in (
+                    inspect.Parameter.POSITIONAL_ONLY,
+                    inspect.Parameter.POSITIONAL_OR_KEYWORD,
+                ):
+                    args_list.append(val)
+                else:
+                    kwargs_dict[name] = val
+            return f(*args_list, **kwargs_dict)
+
+        return wrapper
+
+    return deco