"""Channel Goal Management Tools (v3)
Per-channel long-term goals with sub-task tracking, stored in Redis.
"""
from __future__ import annotations
import jsonutil as json
import uuid
import logging
from datetime import datetime
from typing import Any, Dict, List, Optional, TYPE_CHECKING
from tools.alter_privileges import has_privilege, PRIVILEGES
if TYPE_CHECKING:
from tool_context import ToolContext
logger = logging.getLogger(__name__)
async def _redis(ctx):
"""Resolve the Redis client from the tool context or raise.
Small accessor that every goal handler calls first to obtain the shared
async Redis connection (``ctx.redis``), which is where this module persists
all per-channel goals under ``stargazer:goal:`` keys. Centralizing the lookup
means a missing client fails loudly here rather than producing confusing
``None`` errors deeper in a handler.
Called by the goal handlers in this module (``_create_goal``, ``_get_goal``,
``_list_channel_goals``, ``_update_goal``, ``_delete_goal``, the sub-task
handlers, and ``_list_all_goals``) and by ``get_channel_goals_for_prompt``
when given a ``ctx`` instead of an explicit client; not invoked from outside.
Args:
ctx: Tool execution context providing access to bot internals; its
``redis`` attribute holds the async Redis client.
Returns:
The async Redis client from ``ctx.redis``.
Raises:
RuntimeError: If the context exposes no Redis client.
"""
r = getattr(ctx, "redis", None)
if r is None:
raise RuntimeError("Redis not available")
return r
# ---------------------------------------------------------------
# Authorization
# ---------------------------------------------------------------
async def _check_goal_access(ctx) -> str | None:
"""Gate goal mutations on the ``LONG_TERM_GOALS`` privilege.
Authorization helper guarding every state-changing goal handler. It reads
the caller identity and config off the context and consults
``tools.alter_privileges.has_privilege`` (which checks the Redis-backed
privilege store) so that only users granted ``LONG_TERM_GOALS`` may create,
update, or delete goals and sub-tasks. Read-only handlers deliberately do
not call this.
Called by ``_create_goal``, ``_update_goal``, ``_delete_goal``,
``_add_subtask``, ``_update_subtask``, and ``_remove_subtask`` in this
module; it has no external callers.
Args:
ctx: Tool execution context; supplies ``user_id``, ``redis``, and
``config`` for the privilege check.
Returns:
A JSON error string (``success: false``) when the user lacks the
privilege, or ``None`` when access is allowed and the caller may proceed.
"""
user_id = getattr(ctx, "user_id", "") or ""
redis = getattr(ctx, "redis", None)
config = getattr(ctx, "config", None)
if not await has_privilege(redis, user_id, PRIVILEGES["LONG_TERM_GOALS"], config):
return json.dumps(
{
"success": False,
"error": "The user does not have the LONG_TERM_GOALS privilege. Ask an admin to grant it.",
}
)
return None
# ---------------------------------------------------------------
# Internal helper (also importable by prompt_context.py)
# ---------------------------------------------------------------
[docs]
async def get_channel_goals_for_prompt(
target_channel_id: str,
*,
redis_client=None,
ctx=None,
) -> List[Dict[str, Any]]:
"""Return a simplified, sorted goal list for system-prompt injection.
Reads every goal stored under ``stargazer:goal:<channel>:*`` for the given
channel from Redis (batched through a single pipeline to avoid N-1 round
trips), then flattens each goal into a compact dict with a computed
completion percentage and a ``"<done>/<total> tasks completed"`` summary,
sorted by priority then creation time. The trimmed shape exists so the
runtime prompt assembly can surface active channel goals to the model
without dumping full sub-task payloads. Any error is swallowed and logged,
returning an empty list so prompt rendering never fails on goal data.
Accepts either an explicit ``redis_client`` or a ``ctx`` (from which the
client is resolved via ``_redis``). Called by ``prompt_context.py`` while
assembling runtime prompt context (imported lazily as
``tools.goal_tools.get_channel_goals_for_prompt``); no other callers.
Args:
target_channel_id: Channel whose goals to fetch.
redis_client: Pre-resolved async Redis client; takes precedence over
``ctx`` when provided.
ctx: Tool execution context used to resolve a Redis client when
``redis_client`` is ``None``.
Returns:
A list of simplified goal dicts (goal id, title, description, priority,
completion flag, progress percent, sub-task summary, timestamps),
ordered by priority then creation time; empty if there are no goals, no
client, or an error occurs.
"""
try:
r = redis_client
if r is None and ctx is not None:
r = await _redis(ctx)
if r is None:
return []
pattern = f"stargazer:goal:{target_channel_id}:*"
keys = await r.keys(pattern)
if not keys:
return []
# Batch Redis reads via pipeline (saves N-1 round trips)
pipe = r.pipeline()
for key in keys:
pipe.get(key)
raw_values = await pipe.execute()
goals: list[dict] = []
for raw in raw_values:
if not raw:
continue
goal = json.loads(raw)
subs = goal.get("subtasks", [])
total = len(subs)
done = sum(1 for t in subs if t.get("completed"))
pct = (done / total * 100) if total else 0
goals.append(
{
"goal_id": goal.get("goal_id"),
"title": goal.get("title"),
"description": goal.get("description"),
"priority": goal.get("priority"),
"completed": goal.get("completed", False),
"progress_percent": round(pct, 1),
"subtask_summary": (f"{done}/{total} tasks completed"),
"created_at": goal.get("created_at"),
"updated_at": goal.get("updated_at"),
}
)
prio = {"high": 0, "medium": 1, "low": 2}
goals.sort(
key=lambda g: (
prio.get(g["priority"], 1),
g["created_at"],
)
)
return goals
except Exception as e:
logger.error(
"Error retrieving goals for prompt: %s",
e,
exc_info=True,
)
return []
# ---------------------------------------------------------------
# Tool handlers
# ---------------------------------------------------------------
async def _create_goal(
title: str,
description: str = "",
priority: str = "medium",
ctx: ToolContext | None = None,
) -> str:
"""Create a new long-term goal for the current channel.
Backs the ``create_goal`` tool. After confirming the caller holds
``LONG_TERM_GOALS`` (via ``_check_goal_access``), it mints a UUID goal id,
builds a goal record (with an empty sub-task list and UTC created/updated
timestamps), and persists it to Redis under
``stargazer:goal:<channel>:<goal_id>`` using ``SET``. An invalid priority is
coerced to ``"medium"``.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments, ctx=ctx)`` for the registered ``create_goal``
tool; there are no direct internal callers.
Args:
title (str): Main objective/title of the goal.
description (str): Optional human-readable description.
priority (str): One of ``"low"``, ``"medium"``, or ``"high"``; anything
else is coerced to ``"medium"``.
ctx (ToolContext | None): Tool execution context; supplies the channel
id, Redis client, and privilege inputs.
Returns:
str: JSON. On success ``{"success": true, "goal_id": ..., "goal": ...}``;
on a missing privilege, the error JSON from ``_check_goal_access``.
"""
auth_err = await _check_goal_access(ctx)
if auth_err:
return auth_err
r = await _redis(ctx)
cid = ctx.channel_id
gid = str(uuid.uuid4())
if priority not in ("low", "medium", "high"):
priority = "medium"
goal = {
"goal_id": gid,
"channel_id": cid,
"title": title,
"description": description,
"priority": priority,
"created_at": datetime.utcnow().isoformat(),
"updated_at": datetime.utcnow().isoformat(),
"completed": False,
"subtasks": [],
}
key = f"stargazer:goal:{cid}:{gid}"
await r.set(key, json.dumps(goal))
return json.dumps(
{
"success": True,
"message": f"Goal created for channel {cid}",
"goal_id": gid,
"goal": goal,
}
)
async def _get_goal(
goal_id: str,
ctx: ToolContext | None = None,
) -> str:
"""Fetch a single goal and its progress by id.
Backs the read-only ``get_goal`` tool. Loads the goal record from Redis at
``stargazer:goal:<channel>:<goal_id>`` via ``GET``, then derives total and
completed sub-task counts and a progress percentage to return alongside the
full goal. Requires no privilege, so it intentionally skips
``_check_goal_access``.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments, ctx=ctx)`` for the registered ``get_goal``
tool; there are no direct internal callers.
Args:
goal_id (str): The unique id of the goal to fetch.
ctx (ToolContext | None): Tool execution context; supplies the channel
id and Redis client.
Returns:
str: JSON. On success ``{"success": true, "goal": ..., "progress": ...}``;
``{"success": false, "error": ...}`` when no goal matches the id.
"""
r = await _redis(ctx)
cid = ctx.channel_id
key = f"stargazer:goal:{cid}:{goal_id}"
raw = await r.get(key)
if not raw:
return json.dumps(
{
"success": False,
"error": f"No goal found with ID {goal_id}",
}
)
goal = json.loads(raw)
subs = goal.get("subtasks", [])
total = len(subs)
done = sum(1 for t in subs if t.get("completed"))
return json.dumps(
{
"success": True,
"goal": goal,
"progress": {
"total_subtasks": total,
"completed_subtasks": done,
"progress_percent": ((done / total * 100) if total else 0),
},
}
)
async def _list_channel_goals(
ctx: ToolContext | None = None,
) -> str:
"""List all goals for the current channel with progress summaries.
Backs the read-only ``list_channel_goals`` tool. Scans Redis for
``stargazer:goal:<channel>:*`` keys, loads each goal with ``GET``, and emits
a compact per-goal summary including total/completed sub-task counts and a
progress percentage. Requires no privilege, so it skips
``_check_goal_access``.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments, ctx=ctx)`` for the registered
``list_channel_goals`` tool; there are no direct internal callers.
Args:
ctx (ToolContext | None): Tool execution context; supplies the channel
id and Redis client.
Returns:
str: JSON with ``{"success": true, "channel_id": ..., "goals": [...],
"total_goals": N}``.
"""
r = await _redis(ctx)
cid = ctx.channel_id
keys = await r.keys(f"stargazer:goal:{cid}:*")
goals: list[dict] = []
for key in keys:
raw = await r.get(key)
if not raw:
continue
goal = json.loads(raw)
subs = goal.get("subtasks", [])
total = len(subs)
done = sum(1 for t in subs if t.get("completed"))
goals.append(
{
"goal_id": goal.get("goal_id"),
"title": goal.get("title"),
"description": goal.get("description"),
"priority": goal.get("priority"),
"completed": goal.get("completed", False),
"total_subtasks": total,
"completed_subtasks": done,
"progress_percent": ((done / total * 100) if total else 0),
"created_at": goal.get("created_at"),
"updated_at": goal.get("updated_at"),
}
)
return json.dumps(
{
"success": True,
"channel_id": cid,
"goals": goals,
"total_goals": len(goals),
}
)
async def _update_goal(
goal_id: str,
title: Optional[str] = None,
description: Optional[str] = None,
priority: Optional[str] = None,
completed: Optional[bool] = None,
ctx: ToolContext | None = None,
) -> str:
"""Update a goal's title, description, priority, or completion status.
Backs the ``update_goal`` tool. After the ``LONG_TERM_GOALS`` privilege check
(``_check_goal_access``), it loads the goal from Redis at
``stargazer:goal:<channel>:<goal_id>``, applies whichever fields were
supplied (priority changes are validated against the allowed set), refreshes
``updated_at``, and writes the record back with ``SET``. Fields left as
``None`` are untouched.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments, ctx=ctx)`` for the registered ``update_goal``
tool; there are no direct internal callers.
Args:
goal_id (str): The unique id of the goal to update.
title (Optional[str]): New title, or ``None`` to leave unchanged.
description (Optional[str]): New description, or ``None``.
priority (Optional[str]): New priority; only applied when it is one of
``"low"``, ``"medium"``, or ``"high"``.
completed (Optional[bool]): New completion flag, or ``None``.
ctx (ToolContext | None): Tool execution context; supplies the channel
id, Redis client, and privilege inputs.
Returns:
str: JSON. On success ``{"success": true, "goal": ...}``;
``{"success": false, "error": ...}`` when the goal is missing, or the
privilege error from ``_check_goal_access``.
"""
auth_err = await _check_goal_access(ctx)
if auth_err:
return auth_err
r = await _redis(ctx)
cid = ctx.channel_id
key = f"stargazer:goal:{cid}:{goal_id}"
raw = await r.get(key)
if not raw:
return json.dumps(
{
"success": False,
"error": f"No goal found with ID {goal_id}",
}
)
goal = json.loads(raw)
if title is not None:
goal["title"] = title
if description is not None:
goal["description"] = description
if priority is not None and priority in ("low", "medium", "high"):
goal["priority"] = priority
if completed is not None:
goal["completed"] = completed
goal["updated_at"] = datetime.utcnow().isoformat()
await r.set(key, json.dumps(goal))
return json.dumps(
{
"success": True,
"message": f"Goal {goal_id} updated",
"goal": goal,
}
)
async def _delete_goal(
goal_id: str,
ctx: ToolContext | None = None,
) -> str:
"""Delete a goal and all of its sub-tasks.
Backs the ``delete_goal`` tool. After the ``LONG_TERM_GOALS`` privilege check
(``_check_goal_access``), it issues a Redis ``DELETE`` on
``stargazer:goal:<channel>:<goal_id>``, which removes the goal together with
its embedded sub-tasks (they live inside the same JSON record). The delete
count distinguishes a real removal from a missing id.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments, ctx=ctx)`` for the registered ``delete_goal``
tool; there are no direct internal callers.
Args:
goal_id (str): The unique id of the goal to delete.
ctx (ToolContext | None): Tool execution context; supplies the channel
id, Redis client, and privilege inputs.
Returns:
str: JSON. ``{"success": true, ...}`` when a goal was removed,
``{"success": false, "error": ...}`` when no goal matched, or the
privilege error from ``_check_goal_access``.
"""
auth_err = await _check_goal_access(ctx)
if auth_err:
return auth_err
r = await _redis(ctx)
cid = ctx.channel_id
key = f"stargazer:goal:{cid}:{goal_id}"
deleted = await r.delete(key)
if deleted:
return json.dumps(
{
"success": True,
"message": f"Goal {goal_id} deleted",
}
)
return json.dumps(
{
"success": False,
"error": f"No goal found with ID {goal_id}",
}
)
async def _add_subtask(
goal_id: str,
task_title: str,
task_description: str = "",
ctx: ToolContext | None = None,
) -> str:
"""Append a sub-task to an existing goal.
Backs the ``add_subtask`` tool, breaking a goal into trackable pieces. After
the ``LONG_TERM_GOALS`` privilege check (``_check_goal_access``), it loads
the goal from Redis at ``stargazer:goal:<channel>:<goal_id>``, assigns the
new sub-task a 1-based sequential id (length of the existing list plus one),
appends it with created/completed metadata, bumps ``updated_at``, and writes
the record back with ``SET``.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments, ctx=ctx)`` for the registered ``add_subtask``
tool; there are no direct internal callers.
Args:
goal_id (str): The goal to add the sub-task to.
task_title (str): Title of the new sub-task.
task_description (str): Optional sub-task description.
ctx (ToolContext | None): Tool execution context; supplies the channel
id, Redis client, and privilege inputs.
Returns:
str: JSON. On success ``{"success": true, "subtask": ...}``;
``{"success": false, "error": ...}`` when the goal is missing, or the
privilege error from ``_check_goal_access``.
"""
auth_err = await _check_goal_access(ctx)
if auth_err:
return auth_err
r = await _redis(ctx)
cid = ctx.channel_id
key = f"stargazer:goal:{cid}:{goal_id}"
raw = await r.get(key)
if not raw:
return json.dumps(
{
"success": False,
"error": f"No goal found with ID {goal_id}",
}
)
goal = json.loads(raw)
sid = len(goal.get("subtasks", [])) + 1
subtask = {
"id": sid,
"title": task_title,
"description": task_description,
"completed": False,
"created_at": datetime.utcnow().isoformat(),
"completed_at": None,
}
goal.setdefault("subtasks", []).append(subtask)
goal["updated_at"] = datetime.utcnow().isoformat()
await r.set(key, json.dumps(goal))
return json.dumps(
{
"success": True,
"message": f"Sub-task added to goal {goal_id}",
"subtask": subtask,
}
)
async def _update_subtask(
goal_id: str,
subtask_id: int,
completed: Optional[bool] = None,
task_title: Optional[str] = None,
task_description: Optional[str] = None,
ctx: ToolContext | None = None,
) -> str:
"""Update a sub-task's completion status, title, or description.
Backs the ``update_subtask`` tool. After the ``LONG_TERM_GOALS`` privilege
check (``_check_goal_access``), it loads the parent goal from Redis at
``stargazer:goal:<channel>:<goal_id>``, finds the matching sub-task by id,
and applies whichever fields were supplied; toggling ``completed`` also
stamps or clears ``completed_at``. It refreshes the goal ``updated_at`` and
persists with ``SET``. Fields left as ``None`` are untouched.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments, ctx=ctx)`` for the registered
``update_subtask`` tool; there are no direct internal callers.
Args:
goal_id (str): The goal containing the sub-task.
subtask_id (int): The 1-based id of the sub-task to update.
completed (Optional[bool]): New completion flag; when set, also updates
``completed_at``.
task_title (Optional[str]): New sub-task title, or ``None``.
task_description (Optional[str]): New sub-task description, or ``None``.
ctx (ToolContext | None): Tool execution context; supplies the channel
id, Redis client, and privilege inputs.
Returns:
str: JSON. On success ``{"success": true, "subtask": ...}``;
``{"success": false, "error": ...}`` when the goal or sub-task is
missing, or the privilege error from ``_check_goal_access``.
"""
auth_err = await _check_goal_access(ctx)
if auth_err:
return auth_err
r = await _redis(ctx)
cid = ctx.channel_id
key = f"stargazer:goal:{cid}:{goal_id}"
raw = await r.get(key)
if not raw:
return json.dumps(
{
"success": False,
"error": f"No goal found with ID {goal_id}",
}
)
goal = json.loads(raw)
for sub in goal.get("subtasks", []):
if sub.get("id") == subtask_id:
if completed is not None:
sub["completed"] = completed
sub["completed_at"] = (
datetime.utcnow().isoformat() if completed else None
)
if task_title is not None:
sub["title"] = task_title
if task_description is not None:
sub["description"] = task_description
goal["updated_at"] = datetime.utcnow().isoformat()
await r.set(key, json.dumps(goal))
return json.dumps(
{
"success": True,
"message": (f"Sub-task {subtask_id} updated"),
"subtask": sub,
}
)
return json.dumps(
{
"success": False,
"error": f"Sub-task {subtask_id} not found",
}
)
async def _list_subtasks(
goal_id: str,
ctx: ToolContext | None = None,
) -> str:
"""List a goal's sub-tasks with completion status.
Backs the read-only ``list_subtasks`` tool. Loads the goal from Redis at
``stargazer:goal:<channel>:<goal_id>`` via ``GET`` and returns its full
sub-task list alongside total/completed counts and a progress percentage.
Requires no privilege, so it skips ``_check_goal_access``.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments, ctx=ctx)`` for the registered
``list_subtasks`` tool; there are no direct internal callers.
Args:
goal_id (str): The goal whose sub-tasks to list.
ctx (ToolContext | None): Tool execution context; supplies the channel
id and Redis client.
Returns:
str: JSON. On success ``{"success": true, "subtasks": [...], "total": N,
"completed": M, ...}``; ``{"success": false, "error": ...}`` when the
goal is missing.
"""
r = await _redis(ctx)
cid = ctx.channel_id
key = f"stargazer:goal:{cid}:{goal_id}"
raw = await r.get(key)
if not raw:
return json.dumps(
{
"success": False,
"error": f"No goal found with ID {goal_id}",
}
)
goal = json.loads(raw)
subs = goal.get("subtasks", [])
total = len(subs)
done = sum(1 for t in subs if t.get("completed"))
return json.dumps(
{
"success": True,
"goal_id": goal_id,
"goal_title": goal.get("title"),
"subtasks": subs,
"total": total,
"completed": done,
"progress_percent": ((done / total * 100) if total else 0),
}
)
async def _remove_subtask(
goal_id: str,
subtask_id: int,
ctx: ToolContext | None = None,
) -> str:
"""Permanently remove a sub-task from a goal.
Backs the ``remove_subtask`` tool. After the ``LONG_TERM_GOALS`` privilege
check (``_check_goal_access``), it loads the goal from Redis at
``stargazer:goal:<channel>:<goal_id>``, rebuilds the sub-task list excluding
the target id, and only writes back (with ``SET``, after bumping
``updated_at``) if a sub-task was actually dropped; an unchanged length means
the id was not found.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments, ctx=ctx)`` for the registered
``remove_subtask`` tool; there are no direct internal callers.
Args:
goal_id (str): The goal containing the sub-task.
subtask_id (int): The id of the sub-task to remove.
ctx (ToolContext | None): Tool execution context; supplies the channel
id, Redis client, and privilege inputs.
Returns:
str: JSON. On success ``{"success": true, ...}``;
``{"success": false, "error": ...}`` when the goal or sub-task is
missing, or the privilege error from ``_check_goal_access``.
"""
auth_err = await _check_goal_access(ctx)
if auth_err:
return auth_err
r = await _redis(ctx)
cid = ctx.channel_id
key = f"stargazer:goal:{cid}:{goal_id}"
raw = await r.get(key)
if not raw:
return json.dumps(
{
"success": False,
"error": f"No goal found with ID {goal_id}",
}
)
goal = json.loads(raw)
before = len(goal.get("subtasks", []))
goal["subtasks"] = [
t for t in goal.get("subtasks", []) if t.get("id") != subtask_id
]
if len(goal["subtasks"]) == before:
return json.dumps(
{
"success": False,
"error": f"Sub-task {subtask_id} not found",
}
)
goal["updated_at"] = datetime.utcnow().isoformat()
await r.set(key, json.dumps(goal))
return json.dumps(
{
"success": True,
"message": (f"Sub-task {subtask_id} removed " f"from goal {goal_id}"),
}
)
async def _list_all_goals(
ctx: ToolContext | None = None,
) -> str:
"""List goals across every channel with progress summaries.
Backs the read-only ``list_all_goals`` tool. Scans Redis for all
``stargazer:goal:*`` keys (not scoped to one channel), loads each goal with
``GET``, and emits a compact per-goal summary that includes the owning
``channel_id`` plus total/completed sub-task counts and a progress
percentage.
Dispatched by the tool runner in ``tools/__init__.py`` for the registered
``list_all_goals`` tool (``tool_def.handler(**arguments, ctx=ctx)``), and
also called directly by ``_dump_all_goals`` after its privilege check.
Args:
ctx (ToolContext | None): Tool execution context; supplies the Redis
client.
Returns:
str: JSON with ``{"success": true, "goals": [...], "total_goals": N}``
spanning all channels.
"""
r = await _redis(ctx)
keys = await r.keys("stargazer:goal:*")
goals: list[dict] = []
for key in keys:
raw = await r.get(key)
if not raw:
continue
goal = json.loads(raw)
subs = goal.get("subtasks", [])
total = len(subs)
done = sum(1 for t in subs if t.get("completed"))
goals.append(
{
"channel_id": goal.get("channel_id"),
"goal_id": goal.get("goal_id"),
"title": goal.get("title"),
"description": goal.get("description"),
"priority": goal.get("priority"),
"completed": goal.get("completed", False),
"total_subtasks": total,
"completed_subtasks": done,
"progress_percent": ((done / total * 100) if total else 0),
"created_at": goal.get("created_at"),
"updated_at": goal.get("updated_at"),
}
)
return json.dumps(
{
"success": True,
"goals": goals,
"total_goals": len(goals),
}
)
async def _dump_all_goals(
ctx: ToolContext | None = None,
) -> str:
"""Privileged dump of every goal across all channels.
Backs the ``dump_all_goals`` tool. Unlike ``list_all_goals``, this is gated
on the ``DUMP_GOALS`` privilege (checked via
``tools.alter_privileges.has_privilege`` against the Redis-backed store);
when authorized, it simply delegates to ``_list_all_goals`` to read and
summarize every ``stargazer:goal:*`` record.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments, ctx=ctx)`` for the registered
``dump_all_goals`` tool; there are no direct internal callers.
Args:
ctx (ToolContext | None): Tool execution context; supplies ``user_id``,
``redis``, and ``config`` for the privilege check and the downstream
scan.
Returns:
str: The cross-channel goal listing JSON from ``_list_all_goals`` when
authorized, otherwise a ``{"success": false, "error": ...}`` JSON noting
the missing ``DUMP_GOALS`` privilege.
"""
redis_client = getattr(ctx, "redis", None)
user_id = getattr(ctx, "user_id", "")
config = getattr(ctx, "config", None)
allowed = await has_privilege(
redis_client,
user_id,
PRIVILEGES["DUMP_GOALS"],
config,
)
if not allowed:
return json.dumps(
{
"success": False,
"error": (
"The calling user does not have the "
"DUMP_GOALS privilege. This tool "
"requires DUMP_GOALS to be granted "
"to the invoking user."
),
}
)
return await _list_all_goals(ctx=ctx)
# ---------------------------------------------------------------
# Multi-tool registration
# ---------------------------------------------------------------
TOOLS = [
{
"name": "create_goal",
"description": (
"Create a new long-term goal for the current "
"channel. Each channel can have multiple goals."
),
"parameters": {
"type": "object",
"properties": {
"title": {
"type": "string",
"description": ("Main title/objective of the goal."),
},
"description": {
"type": "string",
"description": ("Detailed description of the goal."),
},
"priority": {
"type": "string",
"enum": ["low", "medium", "high"],
"description": ("Priority level (default: medium)."),
},
},
"required": ["title"],
},
"handler": _create_goal,
},
{
"name": "get_goal",
"description": ("Get a specific goal and its progress " "by goal_id."),
"parameters": {
"type": "object",
"properties": {
"goal_id": {
"type": "string",
"description": ("The unique ID of the goal."),
},
},
"required": ["goal_id"],
},
"handler": _get_goal,
},
{
"name": "list_channel_goals",
"description": (
"List all goals for the current channel " "with progress summaries."
),
"parameters": {
"type": "object",
"properties": {},
},
"handler": _list_channel_goals,
},
{
"name": "update_goal",
"description": (
"Update an existing goal's title, description, "
"priority, or completion status."
),
"parameters": {
"type": "object",
"properties": {
"goal_id": {
"type": "string",
"description": ("The unique ID of the goal " "to update."),
},
"title": {
"type": "string",
"description": "New title.",
},
"description": {
"type": "string",
"description": "New description.",
},
"priority": {
"type": "string",
"enum": ["low", "medium", "high"],
"description": "New priority.",
},
"completed": {
"type": "boolean",
"description": ("Mark goal completed or not."),
},
},
"required": ["goal_id"],
},
"handler": _update_goal,
},
{
"name": "delete_goal",
"description": ("Delete a specific goal and all its " "sub-tasks."),
"parameters": {
"type": "object",
"properties": {
"goal_id": {
"type": "string",
"description": ("The unique ID of the goal " "to delete."),
},
},
"required": ["goal_id"],
},
"handler": _delete_goal,
},
{
"name": "add_subtask",
"description": (
"Add a sub-task to a goal. Sub-tasks help "
"break down goals into manageable pieces."
),
"parameters": {
"type": "object",
"properties": {
"goal_id": {
"type": "string",
"description": ("The goal to add the sub-task to."),
},
"task_title": {
"type": "string",
"description": ("Title of the sub-task."),
},
"task_description": {
"type": "string",
"description": ("Detailed description of the " "sub-task."),
},
},
"required": ["goal_id", "task_title"],
},
"handler": _add_subtask,
},
{
"name": "update_subtask",
"description": (
"Update a sub-task's completion status, " "title, or description."
),
"parameters": {
"type": "object",
"properties": {
"goal_id": {
"type": "string",
"description": ("The goal containing the " "sub-task."),
},
"subtask_id": {
"type": "integer",
"description": ("The ID of the sub-task " "to update."),
},
"completed": {
"type": "boolean",
"description": ("Mark sub-task as completed " "or not."),
},
"task_title": {
"type": "string",
"description": "New title.",
},
"task_description": {
"type": "string",
"description": "New description.",
},
},
"required": ["goal_id", "subtask_id"],
},
"handler": _update_subtask,
},
{
"name": "list_subtasks",
"description": (
"List all sub-tasks for a specific goal " "with completion status."
),
"parameters": {
"type": "object",
"properties": {
"goal_id": {
"type": "string",
"description": ("The goal to list sub-tasks for."),
},
},
"required": ["goal_id"],
},
"handler": _list_subtasks,
},
{
"name": "remove_subtask",
"description": ("Remove a sub-task from a goal permanently."),
"parameters": {
"type": "object",
"properties": {
"goal_id": {
"type": "string",
"description": ("The goal containing the " "sub-task."),
},
"subtask_id": {
"type": "integer",
"description": ("The ID of the sub-task " "to remove."),
},
},
"required": ["goal_id", "subtask_id"],
},
"handler": _remove_subtask,
},
{
"name": "list_all_goals",
"description": (
"List all goals across all channels " "with progress summaries."
),
"parameters": {
"type": "object",
"properties": {},
},
"handler": _list_all_goals,
},
{
"name": "dump_all_goals",
"description": (
"Dump ALL goals and sub-tasks across "
"ALL channels with full details and "
"progress summaries. Requires the "
"DUMP_GOALS privilege."
),
"parameters": {
"type": "object",
"properties": {},
},
"handler": _dump_all_goals,
},
]