"""Modify Stargazer's self.json persona configuration via search-and-replace."""
from __future__ import annotations
import asyncio
import json
import logging
import os
from typing import TYPE_CHECKING
import aiofiles
if TYPE_CHECKING:
from tool_context import ToolContext
logger = logging.getLogger(__name__)
AUTHORIZED_USER_ID = "82303438955753472"
TOOL_NAME = "modify_self_json"
TOOL_DESCRIPTION = (
"Modify the self.json persona configuration file with search and replace. "
"Requires ALL access authorization. ADMIN ONLY."
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"old_text": {
"type": "string",
"description": "The exact text to search for in self.json (must appear exactly once).",
},
"new_text": {
"type": "string",
"description": "The text to replace it with.",
},
},
"required": ["old_text", "new_text"],
}
[docs]
async def run(
old_text: str,
new_text: str,
ctx: ToolContext | None = None,
) -> str:
"""Execute this tool and return the result.
Args:
old_text (str): The old text value.
new_text (str): The new text value.
ctx (ToolContext | None): Tool execution context providing access to bot internals.
Returns:
str: Result string.
"""
user_id = getattr(ctx, "user_id", None) or ""
has_all_access = False
try:
from tools.security import security_manager
if security_manager:
all_policy = security_manager.policies.get("ALL")
if all_policy and user_id in all_policy.allowed_users:
has_all_access = True
except ImportError:
pass
if user_id != AUTHORIZED_USER_ID and not has_all_access:
return "Error: Unauthorized. Only users with ALL access can modify self.json."
try:
self_json_path = os.path.join("prompts", "self.json")
backup_path = os.path.join("prompts", "self.json.backup")
if not await asyncio.to_thread(os.path.exists, self_json_path):
return f"Error: self.json not found at {self_json_path}"
async with aiofiles.open(self_json_path, "r", encoding="utf-8") as f:
current_content = await f.read()
if old_text not in current_content:
return "Error: The specified old_text was not found in self.json"
if current_content.count(old_text) > 1:
return (
f"Error: The old_text appears {current_content.count(old_text)} times "
"in the file. It must appear exactly once for safe replacement."
)
new_content = current_content.replace(old_text, new_text, 1)
try:
json.loads(new_content)
except json.JSONDecodeError as e:
return f"Error: The modified content is not valid JSON. Validation error: {e}"
async with aiofiles.open(backup_path, "w", encoding="utf-8") as f:
await f.write(current_content)
logger.info("Created backup at %s", backup_path)
async with aiofiles.open(self_json_path, "w", encoding="utf-8") as f:
await f.write(new_content)
logger.info("User %s modified self.json successfully", user_id)
old_preview = old_text[:100] + ("..." if len(old_text) > 100 else "")
new_preview = new_text[:100] + ("..." if len(new_text) > 100 else "")
return (
f"Successfully modified self.json.\n\n"
f"**Change made:**\n```\n{old_preview}\n```\n"
f"↓\n```\n{new_preview}\n```\n\n"
f"Backup saved to: {backup_path}\n\n"
f"**Note:** Changes will take effect on the next bot response."
)
except FileNotFoundError:
return "Error: Could not find self.json file"
except PermissionError:
return "Error: Permission denied when accessing self.json"
except Exception as e:
logger.error("Error modifying self.json: %s", e, exc_info=True)
return f"An unexpected error occurred: {e}"