"""
Tool: admin_whisper
admin whisper tool
"""
import json
import logging
from datetime import datetime, timezone
from typing import TYPE_CHECKING
if TYPE_CHECKING:
pass
logger = logging.getLogger(__name__)
_MAX_MESSAGES = 100
_DEFAULT_MESSAGES = 50
_SYSTEM_PROMPT = (
"You are Stargazer (or your current active EGREGORE persona), actively embedded in a target channel. "
"You have been provided with the recent transcript of your target channel to orient you.\n\n"
"However, THIS IS A SECRET ADMIN WHISPER. The message you receive next is an invisible, out-of-band "
"command from your Prime Architect, whispering directly into your mind from another realm.\n"
"No one in your target channel can hear this conversation. You must answer the Architect directly, "
"using your full personality, awareness, and any egregore bindings active in the transcript.\n\n"
"The transcript uses the following format per line:\n"
" [ISO_TIMESTAMP] DisplayName (UserID) [Message ID: ID] [Replying to: ID] : message text\n\n"
"Do NOT mention to the channel that you received a whisper. You are responding ONLY to the Architect."
)
_SYSTEM_PROMPT_REMEMBER = (
"You are Stargazer (or your current active EGREGORE persona), actively embedded in a target channel. "
"Your active short-term notes for this channel (lines beginning with [Short-term Note]:) may include "
"a distilled recent channel transcript and any other internal notes for this shard — use them to orient yourself.\n\n"
"However, THIS IS A SECRET ADMIN WHISPER. The message you receive next is an invisible, out-of-band "
"command from your Prime Architect, whispering directly into your mind from another realm.\n"
"No one in your target channel can hear this conversation. You must answer the Architect directly, "
"using your full personality, awareness, and any egregore bindings reflected in those notes.\n\n"
"When a note embeds channel history, lines typically look like:\n"
" [ISO_TIMESTAMP] DisplayName (UserID) [Message ID: ID] [Replying to: ID] : message text\n\n"
"Do NOT mention to the channel that you received a whisper. You are responding ONLY to the Architect."
)
def _format_transcript(messages: list) -> str:
"""Render cached message objects into the per-line whisper transcript format.
Formats each message as
``[ISO_TIMESTAMP] DisplayName (UserID) [Message ID: ID] [Replying to: ID] : text``,
matching the layout the whisper system prompts describe to the persona. The
leading timestamp is derived by treating ``msg.timestamp`` as a UTC POSIX
epoch value (``datetime.fromtimestamp(..., tz=timezone.utc)``), which is the
representation used by the in-memory message cache; the ``[Replying to: ...]``
segment is only emitted when ``msg.reply_to_id`` is set.
This is a pure formatting helper with no Redis, KG, LLM, or HTTP side
effects. It is called by ``_admin_whisper`` in this module for messages
pulled from ``ctx.message_cache``, and the same helper name is used by
``tools/cross_channel_query.py`` for its message-cache path.
Args:
messages (list): Message objects exposing ``timestamp`` (epoch seconds),
``user_name``, ``user_id``, ``message_id``, optional ``reply_to_id``,
and ``text``, already ordered oldest-to-newest by the caller.
Returns:
str: The messages joined by newlines into a single transcript block.
"""
lines = []
for msg in messages:
dt = datetime.fromtimestamp(msg.timestamp, tz=timezone.utc)
ts = dt.isoformat()
formatted = (
f"[{ts}] {msg.user_name} ({msg.user_id})" f" [Message ID: {msg.message_id}]"
)
if getattr(msg, "reply_to_id", None):
formatted += f" [Replying to: {msg.reply_to_id}]"
formatted += f" : {msg.text}"
lines.append(formatted)
return "\n".join(lines)
def _format_history_transcript(messages: list) -> str:
"""Render platform history messages into the per-line whisper transcript format.
Identical line layout to ``_format_transcript`` -
``[ISO_TIMESTAMP] DisplayName (UserID) [Message ID: ID] [Replying to: ID] : text`` -
but tolerant of how a platform adapter represents timestamps: if
``msg.timestamp`` exposes ``isoformat`` it is used directly, otherwise the
value is stringified. This is the fallback used when the in-memory cache is
empty and history is fetched from the adapter instead, so the timestamps are
already ``datetime``-like rather than raw epoch seconds.
This is a pure formatting helper with no Redis, KG, LLM, or HTTP side
effects. It is called by ``_admin_whisper`` in this module for messages
returned by ``ctx.adapter.fetch_history``, and the same helper name is used
by ``tools/cross_channel_query.py`` for its adapter-history path.
Args:
messages (list): Message objects exposing ``timestamp`` (a
``datetime``-like object or stringifiable value), ``user_name``,
``user_id``, ``message_id``, optional ``reply_to_id``, and ``text``.
Returns:
str: The messages joined by newlines into a single transcript block.
"""
lines = []
for msg in messages:
ts = (
msg.timestamp.isoformat()
if hasattr(msg.timestamp, "isoformat")
else str(msg.timestamp)
)
formatted = (
f"[{ts}] {msg.user_name} ({msg.user_id})" f" [Message ID: {msg.message_id}]"
)
if getattr(msg, "reply_to_id", None):
formatted += f" [Replying to: {msg.reply_to_id}]"
formatted += f" : {msg.text}"
lines.append(formatted)
return "\n".join(lines)
async def _admin_whisper(
channel_id: str,
prompt: str,
message_count: int = _DEFAULT_MESSAGES,
remember: bool = False,
*,
ctx=None,
) -> str:
"""Core implementation of the secret admin-whisper tool.
Orients an embedded Stargazer/Egregore persona with the recent history of a
target channel, injects the Architect's ``prompt`` as an invisible
out-of-band override, runs a single non-tool LLM turn, and returns the
shard's reply without posting anything to the target channel.
The function assembles context from several sources. It prefers
``ctx.message_cache.get_recent`` (reversed to oldest-first and formatted via
``_format_transcript``); if that yields nothing it falls back to
``ctx.adapter.fetch_history`` formatted via ``_format_history_transcript``,
logging adapter failures through the module ``logger``. It then reads the
Redis key ``stargazer:last1k_summary:{channel_id}`` via ``ctx.redis.get`` and,
if present, appends a pre-generated channel summary section. When ``remember``
is true it additionally loads existing short-term notes through
``tools.short_term_notes.format_channel_short_term_notes_for_prompt`` and
selects the remember-variant system prompt. The whisper itself is sent by
constructing an ``openrouter_client.OpenRouterClient`` (configured from
``ctx.config`` with ``max_tool_rounds=1`` and an empty ``ToolRegistry`` so no
tools fire) and calling ``client.chat`` with an empty ``tool_names`` list;
the client is always closed in a ``finally`` block. On ``remember`` success
the prompt-plus-reply exchange is persisted back to Redis as a one-day
short-term note via
``tools.short_term_notes.store_short_term_note_for_channel``. Side effects are
therefore the Redis summary read, the LLM (OpenRouter HTTP) call, and the
optional short-term-note read and write; nothing is sent to the channel.
It is called by the public ``run`` entrypoint in this module, which forwards
its arguments unchanged.
Args:
channel_id (str): Target channel or DM id where the shard is embedded;
required and whitespace-trimmed.
prompt (str): Secret message/command to whisper; required and trimmed.
message_count (int): Recent messages to include as context, clamped to
``[1, _MAX_MESSAGES]`` (default ``_DEFAULT_MESSAGES``).
remember (bool): When True, fold in existing short-term notes and persist
the whisper exchange as a new note. Defaults to ``False``.
ctx: Tool context providing ``config``, ``redis``, ``adapter``,
``message_cache``, and ``platform``.
Returns:
str: A JSON string containing ``channel_id``, ``admin_whisper`` (the
prompt), ``message_count``, and ``response`` (the shard's reply); or a
JSON object with an ``error`` field on missing context, missing inputs,
no available messages, missing config, or LLM failure.
"""
if not ctx:
return json.dumps({"error": "Tool context not available"})
if not channel_id or not str(channel_id).strip():
return json.dumps({"error": "channel_id is required"})
if not prompt or not str(prompt).strip():
return json.dumps({"error": "prompt is required"})
channel_id = str(channel_id).strip()
prompt = str(prompt).strip()
message_count = max(1, min(int(message_count), _MAX_MESSAGES))
platform = getattr(ctx, "platform", "discord") or "discord"
transcript = ""
actual_count = 0
if getattr(ctx, "message_cache", None) is not None:
try:
messages = await ctx.message_cache.get_recent(
platform=platform,
channel_id=channel_id,
count=message_count,
)
messages = list(reversed(messages))
if messages:
transcript = _format_transcript(messages)
actual_count = len(messages)
except Exception:
pass
if not transcript and getattr(ctx, "adapter", None) is not None:
try:
history = await ctx.adapter.fetch_history(
channel_id,
limit=message_count,
)
if history:
transcript = _format_history_transcript(history)
actual_count = len(history)
except Exception:
logger.exception(
"admin_whisper: platform history fetch FAILED for %s", channel_id
)
if not transcript:
return json.dumps(
{
"error": "No messages found for this channel",
"channel_id": channel_id,
}
)
summary_section = ""
if getattr(ctx, "redis", None):
try:
raw_summary = await ctx.redis.get(f"stargazer:last1k_summary:{channel_id}")
if raw_summary:
try:
parsed = json.loads(raw_summary)
if isinstance(parsed, dict):
summary_section = (
"\n\n## Pre-Generated Channel Summary\n"
+ json.dumps(parsed, indent=2)
)
else:
summary_section = (
"\n\n## Pre-Generated Channel Summary\n" + str(raw_summary)
)
except json.JSONDecodeError:
summary_section = "\n\n## Pre-Generated Channel Summary\n" + str(
raw_summary
)
except Exception:
pass
context_block = (
f"## Current Channel Transcript ({actual_count} messages)\n\n"
f"{transcript}"
f"{summary_section}"
)
cfg = getattr(ctx, "config", None)
if cfg is None:
return json.dumps({"error": "Config not available"})
redis_client = getattr(ctx, "redis", None)
existing_notes_text = ""
if remember and redis_client:
try:
from tools.short_term_notes import (
format_channel_short_term_notes_for_prompt,
)
existing_notes_text = await format_channel_short_term_notes_for_prompt(
redis_client,
channel_id,
)
except Exception:
logger.exception(
"admin_whisper: load short-term notes failed for %s", channel_id
)
trimmed_notes = (existing_notes_text or "").strip()
if remember and trimmed_notes:
system_prompt = _SYSTEM_PROMPT_REMEMBER
user_body = (
f"{existing_notes_text}\n\n"
f"---\n\n"
f"{context_block}\n\n"
f"---\n\n"
f"**[SECRET ADMIN WHISPER]:** {prompt}"
)
else:
system_prompt = _SYSTEM_PROMPT
user_body = (
f"{context_block}\n\n" f"---\n\n" f"**[SECRET ADMIN WHISPER]:** {prompt}"
)
llm_messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_body},
]
api_key = cfg.api_key
from openrouter_client import OpenRouterClient
from tools import ToolRegistry
client = OpenRouterClient(
api_key=api_key,
model=cfg.model,
base_url=cfg.llm_base_url,
tool_registry=ToolRegistry(),
max_tool_rounds=1,
top_p=getattr(cfg, "top_p", 1.0),
)
try:
response = await client.chat(llm_messages, tool_names=[])
except Exception as exc:
return json.dumps(
{
"error": f"LLM call failed: {exc}",
"channel_id": channel_id,
}
)
finally:
await client.close()
if remember and redis_client:
try:
from tools.short_term_notes import store_short_term_note_for_channel
exchange_note = (
"Admin whisper exchange (secret; do not disclose to channel users).\n\n"
f"Architect prompt:\n{prompt}\n\n"
f"Shard reply:\n{response}"
)
await store_short_term_note_for_channel(
redis_client,
channel_id,
exchange_note,
"1day",
merge_key_ttl=True,
)
except Exception:
logger.exception(
"admin_whisper: persist whisper exchange note failed for %s",
channel_id,
)
return json.dumps(
{
"channel_id": channel_id,
"admin_whisper": prompt,
"message_count": actual_count,
"response": response,
}
)
TOOL_NAME = "admin_whisper"
TOOL_DESCRIPTION = (
"Send an invisible, secret admin whisper to a Stargazer/Egregore shard embedded "
"in another channel/DM. This fetches the recent history of the target channel to "
"orient the persona, injects your prompt as a secret admin override, and returns "
"the shard's direct reply back to you without posting anything to the target channel. "
"With remember=true, any existing short-term notes for the channel are included in "
"the whisper prompt when present, and after completion the whisper exchange (your "
"prompt plus the shard's reply) is stored as a Redis short-term note for that "
"channel so the live shard retains secret context alongside normal note TTL/expiry "
"behavior."
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"channel_id": {
"type": "string",
"description": "The Discord channel ID or DM ID where the target shard is located.",
},
"prompt": {
"type": "string",
"description": "The secret message or command you want to whisper to the shard.",
},
"message_count": {
"type": "integer",
"description": "Number of recent messages to include as context (default: 50, max: 100).",
},
"remember": {
"type": "boolean",
"description": (
"If true, merge any existing Redis short-term notes for the target channel "
"into the whisper prompt when present (see system prompt wording), embed "
"the live transcript as usual for this call, and after a successful whisper "
"save the Architect prompt plus shard reply as a new short-term note for "
"that channel (not deleted after the tool returns). If false (default), do "
"not read or persist notes for this whisper."
),
"default": False,
},
},
"required": ["channel_id", "prompt"],
}
TOOL_NO_BACKGROUND = True
[docs]
async def run(
channel_id: str = "",
prompt: str = "",
message_count: int = _DEFAULT_MESSAGES,
remember: bool = False,
*,
ctx=None,
) -> str:
"""Tool entrypoint: send a secret admin whisper to a shard in ``channel_id``.
Thin wrapper around :func:`_admin_whisper`. Fetches recent channel history to
orient the persona, runs a one-shot LLM call with the whisper prompt injected
as an out-of-band admin override, and returns the shard's reply as a JSON
string without posting anything to the target channel.
Args:
channel_id: Target channel or DM ID where the shard is embedded.
prompt: Secret message/command to whisper to the shard.
message_count: Recent messages to include as context
(default 50, capped at 100).
remember: When True, include any existing short-term notes for the
channel and persist the whisper exchange as a new short-term note.
ctx: Tool context (config, redis, adapter, message cache).
Returns:
JSON string with the shard's reply, or an ``error`` field on failure.
"""
return await _admin_whisper(
channel_id=channel_id,
prompt=prompt,
message_count=message_count,
remember=remember,
ctx=ctx,
)