"""Channel Goal Management Tools (v3)
Per-channel long-term goals with sub-task tracking, stored in Redis.
"""
from __future__ import annotations
import json
import uuid
import logging
from datetime import datetime
from typing import Any, Dict, List, Optional, TYPE_CHECKING
if TYPE_CHECKING:
from tool_context import ToolContext
logger = logging.getLogger(__name__)
async def _redis(ctx):
"""Internal helper: redis.
Args:
ctx: Tool execution context providing access to bot internals.
"""
r = getattr(ctx, "redis", None)
if r is None:
raise RuntimeError("Redis not available")
return r
# ---------------------------------------------------------------
# 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 simplified goal list for system prompt injection.
Accepts either an explicit *redis_client* or a *ctx*.
"""
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:
"""Internal helper: create goal.
Args:
title (str): The title value.
description (str): Human-readable description.
priority (str): The priority value.
ctx (ToolContext | None): Tool execution context providing access to bot internals.
Returns:
str: Result string.
"""
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:
"""Internal helper: get goal.
Args:
goal_id (str): The goal id value.
ctx (ToolContext | None): Tool execution context providing access to bot internals.
Returns:
str: Result string.
"""
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:
"""Internal helper: list channel goals.
Args:
ctx (ToolContext | None): Tool execution context providing access to bot internals.
Returns:
str: Result string.
"""
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:
"""Internal helper: update goal.
Args:
goal_id (str): The goal id value.
title (Optional[str]): The title value.
description (Optional[str]): Human-readable description.
priority (Optional[str]): The priority value.
completed (Optional[bool]): The completed value.
ctx (ToolContext | None): Tool execution context providing access to bot internals.
Returns:
str: Result string.
"""
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:
"""Internal helper: delete goal.
Args:
goal_id (str): The goal id value.
ctx (ToolContext | None): Tool execution context providing access to bot internals.
Returns:
str: Result string.
"""
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:
"""Internal helper: add subtask.
Args:
goal_id (str): The goal id value.
task_title (str): The task title value.
task_description (str): The task description value.
ctx (ToolContext | None): Tool execution context providing access to bot internals.
Returns:
str: Result string.
"""
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:
"""Internal helper: update subtask.
Args:
goal_id (str): The goal id value.
subtask_id (int): The subtask id value.
completed (Optional[bool]): The completed value.
task_title (Optional[str]): The task title value.
task_description (Optional[str]): The task description value.
ctx (ToolContext | None): Tool execution context providing access to bot internals.
Returns:
str: Result string.
"""
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:
"""Internal helper: list subtasks.
Args:
goal_id (str): The goal id value.
ctx (ToolContext | None): Tool execution context providing access to bot internals.
Returns:
str: Result string.
"""
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:
"""Internal helper: remove subtask.
Args:
goal_id (str): The goal id value.
subtask_id (int): The subtask id value.
ctx (ToolContext | None): Tool execution context providing access to bot internals.
Returns:
str: Result string.
"""
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:
"""Internal helper: list all goals.
Args:
ctx (ToolContext | None): Tool execution context providing access to bot internals.
Returns:
str: Result string.
"""
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 all goals across all channels."""
from tools.alter_privileges import has_privilege, PRIVILEGES
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,
},
]