"""Debug tool: force a limbic inhale for a specific channel.
Initialises the global and local Redis keys if missing and returns
the generated state vector.
"""
from __future__ import annotations
import json
import logging
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from tool_context import ToolContext
logger = logging.getLogger(__name__)
TOOL_NAME = "debug_limbic_shard"
TOOL_DESCRIPTION = (
"Forces a 'Limbic Inhale' for the specified channel. "
"Initializes the Global and Local Redis keys if missing. "
"Returns the generated state vector."
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"channel_id": {
"type": "string",
"description": "The channel/shard ID to inhale for.",
},
},
"required": ["channel_id"],
}
async def _get_db12_client(ctx):
"""Internal helper: get db12 client.
Args:
ctx: Tool execution context providing access to bot internals.
"""
import redis.asyncio as aioredis
r = getattr(ctx, "redis", None)
if r is None:
return None
pool = r.connection_pool
kwargs = pool.connection_kwargs.copy()
kwargs["db"] = 12
return aioredis.Redis(connection_pool=aioredis.ConnectionPool(
connection_class=pool.connection_class, **kwargs,
))
[docs]
async def run(channel_id: str, ctx: "ToolContext | None" = None) -> str:
"""Execute this tool and return the result.
Args:
channel_id (str): Discord/Matrix channel identifier.
ctx ('ToolContext | None'): Tool execution context providing access to bot internals.
Returns:
str: Result string.
"""
try:
from limbic_system import LimbicSystem
db12 = await _get_db12_client(ctx) if ctx else None
if db12 is None:
return "Error: Redis not available via ctx."
limbic = LimbicSystem(redis_client=db12)
state = await limbic.inhale(channel_id)
global_exists = bool(await db12.exists("db12:global"))
shard_exists = bool(await db12.exists(f"db12:shard:{channel_id}"))
return json.dumps(
{
"status": "Inhale Complete",
"state": state,
"persistence_check": {
"db12:global": global_exists,
f"db12:shard:{channel_id}": shard_exists,
},
},
indent=2,
)
except Exception as e:
return f"Critical Failure in Limbic System: {e}"