"""Core-memory editing tools. Thin wrappers over src.memory.core that translate its raises/returns into the dict-return contract every ADK tool follows. """ from google.adk.tools.tool_context import ToolContext from src.memory import core _VALID_SECTIONS = ("persona", "human") def core_memory_append(section: str, content: str, tool_context: ToolContext) -> dict: """Append a fact to core memory as a new bullet line. Use 'human' for stable facts about the user (name, role, city, tools they use). Use 'persona' for stable facts about yourself. Args: section: Which core memory block. Must be 'persona' or 'human'. content: The fact to append, as a short natural-language sentence. Returns: Success: ``{"status": "success", "section": "human", "content_added": "..."}`` Error: ``{"status": "error", "error_message": "..."}`` """ if section not in _VALID_SECTIONS: return { "status": "error", "error_message": f"section must be 'persona' or 'human', got {section!r}", } trimmed = content.strip() if content else "" if not trimmed: return { "status": "error", "error_message": "content must be a non-empty string", } core.append(tool_context.state, section, trimmed) return {"status": "success", "section": section, "content_added": trimmed} def core_memory_replace( section: str, old: str, new: str, tool_context: ToolContext ) -> dict: """Replace an existing entry in core memory. Requires user confirmation. Use this to correct a fact already in memory. The user is shown the proposed change and must approve it before it takes effect. Args: section: Which core memory block. Must be 'persona' or 'human'. old: The exact substring currently in the block, to be replaced. new: The replacement text. Returns: First call: ``{"status": "pending", "message": "..."}`` After approval: ``{"status": "success", "section": ..., "old": ..., "new": ...}`` After rejection: ``{"status": "rejected", "message": "..."}`` On error: ``{"status": "error", "error_message": "..."}`` """ if section not in _VALID_SECTIONS: return { "status": "error", "error_message": f"section must be 'persona' or 'human', got {section!r}", } if not old or not new: return { "status": "error", "error_message": "both 'old' and 'new' must be non-empty", } # Scenario 1: first call — pause and ask if not tool_context.tool_confirmation: tool_context.request_confirmation( hint=f"Replace in {section}: {old!r} → {new!r}. Proceed?", payload={"section": section, "old": old, "new": new}, ) return { "status": "pending", "message": f"Awaiting confirmation to replace {old!r} in {section}", } # Scenario 2: resumed with approval if tool_context.tool_confirmation.confirmed: replaced = core.replace(tool_context.state, section, old, new) if not replaced: return { "status": "error", "error_message": f"{old!r} not found in {section}", } return {"status": "success", "section": section, "old": old, "new": new} # Scenario 3: resumed with rejection return {"status": "rejected", "message": "User declined the edit"}