"""Tool for retrieving full tool call execution traces from Redis."""
import jsonutil as json
import logging
logger = logging.getLogger(__name__)
TOOL_NAME = "retrieve_tool_call_log"
TOOL_DESCRIPTION = (
"Retrieve the full tool call execution details for a specific turn, "
"given a Stargazer-System-Log summary ID, or retrieve a specific tool "
"execution trace given a toolcall ID. Both formats accept prefixes ("
"e.g. 'toolcall_summary:xyz' or 'toolcall:xyz') or bare UUIDs."
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"summary_id": {
"type": "string",
"description": "The UUID of the Stargazer-System-Log summary message (e.g. from :: toolcall:uuid)",
},
},
"required": ["summary_id"],
}
[docs]
async def run(summary_id: str, ctx=None) -> str:
"""Execute this tool and return the result.
Args:
summary_id (str): The UUID of the tool-call summary.
ctx: Tool execution context providing access to bot internals.
Returns:
str: JSON string containing the full execution traces.
"""
if not summary_id:
return json.dumps({"error": "summary_id is required"})
if not ctx or not ctx.message_cache:
return json.dumps({"error": "Redis context not available for retrieval"})
try:
def decode_redis_result(data):
"""Recursively decode raw Redis reply bytes into native Python text.
Walks an arbitrarily nested Redis reply -- a ``dict`` from
``hgetall``, a ``list``, raw ``bytes``, or an already-decoded scalar
-- and returns the same structure with every ``bytes`` value decoded
as UTF-8. Dictionary keys are decoded as well so the resulting
mapping can be queried with plain string keys (e.g.
``results.get("channel_id")``) regardless of whether the Redis
client returned ``str`` or ``bytes``.
This is a pure, local helper that performs no I/O of its own; it is
defined inside :func:`run` and invoked only there -- on the
``hgetall`` replies fetched from ``ctx.message_cache.redis_client``
for the ``toolcall:`` and bare-UUID lookup paths. It has no callers
outside this module.
Args:
data: A Redis reply fragment: a ``dict``, ``list``, ``bytes``,
or any already-native scalar value.
Returns:
The same value with all ``bytes`` (and ``bytes`` dict keys)
decoded to UTF-8 ``str``; non-bytes scalars are returned
unchanged.
"""
if isinstance(data, dict):
return {
decode_redis_result(k): decode_redis_result(v)
for k, v in data.items()
}
if isinstance(data, bytes):
return data.decode("utf-8")
if isinstance(data, list):
return [decode_redis_result(i) for i in data]
return data
summary_id = summary_id.strip()
# Group 1: Explicit toolcall_summary
if summary_id.startswith("toolcall_summary:"):
bare_id = summary_id[len("toolcall_summary:") :]
records = await ctx.message_cache.get_tool_call_records_by_summary(bare_id)
if not records:
return json.dumps({"error": f"No summary found for {summary_id}"})
if records and (
records[0].get("channel_id") != ctx.channel_id
or records[0].get("platform") != ctx.platform
):
return json.dumps(
{
"error": "Unauthorized: These tool logs belong to a different channel."
}
)
return json.dumps(records, indent=2)
# Group 2: Explicit toolcall (single)
elif summary_id.startswith("toolcall:"):
res_raw = await ctx.message_cache.redis_client.hgetall(summary_id)
if not res_raw:
return json.dumps({"error": f"Record {summary_id} not found"})
results = decode_redis_result(res_raw)
if (
results.get("channel_id") != ctx.channel_id
or results.get("platform") != ctx.platform
):
logger.warning(
"Unauthorized attempt to access tool log %s from %s:%s (owner: %s:%s)",
summary_id,
ctx.platform,
ctx.channel_id,
results.get("platform"),
results.get("channel_id"),
)
return json.dumps(
{
"error": "Unauthorized: This tool log belongs to a different channel."
}
)
return json.dumps(results, indent=2)
# Group 3: Bare UUID (Fallback logic)
else:
# Try as summary ID first
records = await ctx.message_cache.get_tool_call_records_by_summary(
summary_id
)
if records:
if (
records[0].get("channel_id") != ctx.channel_id
or records[0].get("platform") != ctx.platform
):
return json.dumps(
{
"error": "Unauthorized: These tool logs belong to a different channel."
}
)
return json.dumps(records, indent=2)
# Fall back to single record ID
full_record_id = f"toolcall:{summary_id}"
res_raw = await ctx.message_cache.redis_client.hgetall(full_record_id)
if res_raw:
results = decode_redis_result(res_raw)
if (
results.get("channel_id") != ctx.channel_id
or results.get("platform") != ctx.platform
):
return json.dumps(
{
"error": "Unauthorized: This tool log belongs to a different channel."
}
)
return json.dumps(results, indent=2)
return json.dumps(
{"error": f"No logs found for summary/record ID {summary_id}"}
)
except Exception as exc:
logger.error(f"Failed to retrieve tool call log for {summary_id}: {exc}")
return json.dumps({"error": f"Internal error during retrieval: {str(exc)}"})