bryanb
/
CheddahBot
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge refactor/multi-agent-framework 

Multi-agent architecture with markdown skills, memory scoping,
cross-agent delegation, and foundation cleanup (thread safety,
file attachments, proper tool message roles, shell approval).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
cora-start
PeninsulaInd 2026-02-17 10:18:10 -06:00
parent a7171673fc 8a21990ba4
commit d376420025
39 changed files with 1159 additions and 560 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
CLAUDE.md
View File

@ -18,13 +18,21 @@ The bot polls ClickUp for tasks, maps them to skills, and auto-executes or asks
```
Gradio UI (ui.py)
↓
Agent (agent.py) ← Memory (memory.py, 4-layer: identity/long-term/daily/semantic)
AgentRegistry (agent_registry.py)
├── default agent ← AgentConfig (config.py)
├── writer agent
├── researcher agent
└── ops agent
↓
Agent (agent.py) ← Memory (memory.py, 4-layer, per-agent scoping)
↓ ← Skills (skills.py, markdown skills with frontmatter)
LLM Adapter (llm.py)
├── Chat brain: OpenRouter / Ollama / LM Studio
├── Chat brain: OpenRouter / Ollama / LM Studio (per-agent model override)
└── Execution brain: Claude Code CLI (subprocess)
↓
Tool Registry (tools/__init__.py) ← auto-discovers tools in tools/
├── delegate_task → execution brain
└── delegate_to_agent → cross-agent delegation (depth-limited)
↓
Scheduler (scheduler.py)
├── Poll loop: cron-based scheduled tasks
@ -40,7 +48,7 @@ NotificationBus (notifications.py) → Gradio / future Discord / Slack
# Run the app
uv run python -m cheddahbot
# Run tests (118 tests, ~3s)
# Run tests (124 tests)
uv run pytest
# Run tests verbose
@ -66,29 +74,35 @@ uv add --group test <package>
| File | Purpose |
|------|---------|
| `cheddahbot/__main__.py` | Entry point, wires all components |
| `cheddahbot/__main__.py` | Entry point, multi-agent wiring |
| `cheddahbot/agent.py` | Core agentic loop (chat + tool execution) |
| `cheddahbot/agent_registry.py` | Multi-agent registry (named agents, default) |
| `cheddahbot/llm.py` | Two-brain LLM adapter |
| `cheddahbot/config.py` | Dataclass config (env → YAML → defaults) |
| `cheddahbot/config.py` | Config + AgentConfig dataclasses |
| `cheddahbot/db.py` | SQLite persistence (WAL, thread-safe) |
| `cheddahbot/scheduler.py` | Three daemon threads: poll, heartbeat, ClickUp |
| `cheddahbot/clickup.py` | ClickUp REST API v2 client (httpx) |
| `cheddahbot/notifications.py` | UI-agnostic pub/sub notification bus |
| `cheddahbot/memory.py` | 4-layer memory with semantic search |
| `cheddahbot/memory.py` | 4-layer memory with semantic search + scoping |
| `cheddahbot/router.py` | System prompt builder |
| `cheddahbot/skills.py` | Markdown skill registry (discovers skills/*.md) |
| `cheddahbot/ui.py` | Gradio web interface |
| `cheddahbot/tools/` | Tool modules (auto-discovered) |
| `config.yaml` | Runtime configuration |
| `cheddahbot/tools/delegate.py` | delegate_task + delegate_to_agent tools |
| `config.yaml` | Runtime configuration (incl. agents section) |
| `identity/SOUL.md` | Agent personality |
| `identity/USER.md` | User profile |
| `skills/` | Prompt templates for tools (press releases, etc.) |
| `skills/` | Markdown skill files with YAML frontmatter |
## Conventions
- **Config precedence**: env vars > config.yaml > dataclass defaults
- **ClickUp env vars**: `CLICKUP_API_TOKEN`, `CLICKUP_WORKSPACE_ID`, `CLICKUP_SPACE_ID`
- **Tool registration**: Use the `@tool("name", "description", category="cat")` decorator in any file under `cheddahbot/tools/` — auto-discovered on startup
- **Tool context**: Tools can accept `ctx: dict = None` to get `config`, `db`, `agent`, `memory` injected
- **Tool context**: Tools can accept `ctx: dict | None = None` to get `config`, `db`, `agent`, `memory`, `agent_registry` injected
- **Skills**: `.md` files in `skills/` with YAML frontmatter (`name`, `description`, `tools`, `agents`). Files without frontmatter are data files (skipped by registry)
- **Multi-agent**: Configure agents in `config.yaml` under `agents:` key. Each agent has `name`, `display_name`, `model` (override), `tools` (whitelist), `memory_scope`. First agent is the default. Use `delegate_to_agent` tool for cross-agent delegation (depth limit: 3).
- **Memory scoping**: Agents with `memory_scope` set use `memory/{scope}/` subdirectory. Empty scope = shared `memory/` root. Fallback search checks both scoped and shared directories.
- **Database**: SQLite with WAL mode, thread-local connections via `threading.local()`
- **KV store**: Task state stored as JSON at `clickup:task:{id}:state` keys
- **ClickUp field mapping**: `Work Category` field (not `Task Type`) identifies task types like "Press Release", "Link Building". The `Client` field (not `Company`) holds the client name.

97
cheddahbot/__main__.py
View File

@ -1,12 +1,12 @@
"""Entry point: python -m cheddahbot"""
import logging
import sys
from .agent import Agent
from .agent_registry import AgentRegistry
from .config import load_config
from .db import Database
from .llm import LLMAdapter
from .agent import Agent
from .ui import create_ui
logging.basicConfig(
@ -26,64 +26,117 @@ def main():
log.info("Chat brain model: %s", config.chat_model)
log.info("Execution brain model: %s (Claude Code CLI)", config.default_model)
llm = LLMAdapter(
default_llm = LLMAdapter(
default_model=config.chat_model,
openrouter_key=config.openrouter_api_key,
ollama_url=config.ollama_url,
lmstudio_url=config.lmstudio_url,
)
if llm.is_execution_brain_available():
if default_llm.is_execution_brain_available():
log.info("Execution brain: Claude Code CLI found in PATH")
else:
log.warning("Execution brain: Claude Code CLI NOT found — heartbeat/scheduler tasks will fail")
log.warning(
"Execution brain: Claude Code CLI NOT found — heartbeat/scheduler tasks will fail"
)
log.info("Creating agent...")
agent = Agent(config, db, llm)
# Phase 2+: Memory system
# Skill registry (markdown skills from skills/ directory)
skills_registry = None
try:
from .memory import MemorySystem
log.info("Initializing memory system...")
memory = MemorySystem(config, db)
agent.set_memory(memory)
except Exception as e:
log.warning("Memory system not available: %s", e)
from .skills import SkillRegistry
# Phase 3+: Tool system
log.info("Initializing skill registry...")
skills_registry = SkillRegistry(config.skills_dir)
log.info("Loaded %d skills", len(skills_registry.list_skills()))
except Exception as e:
log.warning("Skill registry not available: %s", e)
# Tool system (shared across all agents — tools are singletons)
tools = None
try:
from .tools import ToolRegistry
# Create a temporary default agent for tool discovery; will be replaced below
_bootstrap_agent = Agent(config, db, default_llm)
log.info("Initializing tool system...")
tools = ToolRegistry(config, db, agent)
agent.set_tools(tools)
tools = ToolRegistry(config, db, _bootstrap_agent)
except Exception as e:
log.warning("Tool system not available: %s", e)
# Multi-agent setup
registry = AgentRegistry()
log.info("Configuring %d agent(s)...", len(config.agents))
for i, agent_cfg in enumerate(config.agents):
# Per-agent LLM (if model override set)
if agent_cfg.model:
agent_llm = LLMAdapter(
default_model=agent_cfg.model,
openrouter_key=config.openrouter_api_key,
ollama_url=config.ollama_url,
lmstudio_url=config.lmstudio_url,
)
else:
agent_llm = default_llm
agent = Agent(config, db, agent_llm, agent_config=agent_cfg)
# Memory system (with optional scoping)
try:
from .memory import MemorySystem
memory = MemorySystem(config, db, scope=agent_cfg.memory_scope)
agent.set_memory(memory)
except Exception as e:
log.warning("Memory system not available for agent '%s': %s", agent_cfg.name, e)
# Wire shared tool registry and skills
if tools:
agent.set_tools(tools)
if skills_registry:
agent.set_skills_registry(skills_registry)
registry.register(agent_cfg.name, agent, is_default=(i == 0))
log.info(
" Agent '%s' (%s) — tools: %s, scope: %s",
agent_cfg.name,
agent_cfg.display_name,
"all" if agent_cfg.tools is None else str(len(agent_cfg.tools)),
agent_cfg.memory_scope or "shared",
)
# Update tool registry to reference the default agent and agent registry
default_agent = registry.default
if tools and default_agent:
tools.agent = default_agent
tools.agent_registry = registry
# Notification bus (UI-agnostic)
notification_bus = None
try:
from .notifications import NotificationBus
log.info("Initializing notification bus...")
notification_bus = NotificationBus(db)
except Exception as e:
log.warning("Notification bus not available: %s", e)
# Phase 3+: Scheduler
# Scheduler (uses default agent)
try:
from .scheduler import Scheduler
log.info("Starting scheduler...")
scheduler = Scheduler(config, db, agent, notification_bus=notification_bus)
scheduler = Scheduler(config, db, default_agent, notification_bus=notification_bus)
scheduler.start()
except Exception as e:
log.warning("Scheduler not available: %s", e)
log.info("Launching Gradio UI on %s:%s...", config.host, config.port)
app, css = create_ui(agent, config, llm, notification_bus=notification_bus)
app = create_ui(default_agent, config, default_llm, notification_bus=notification_bus)
app.launch(
server_name=config.host,
server_port=config.port,
pwa=True,
show_error=True,
css=css,
)

148
cheddahbot/agent.py
View File

@ -2,12 +2,14 @@
from __future__ import annotations
import base64
import json
import logging
import uuid
from typing import Generator
from collections.abc import Generator
from pathlib import Path
from .config import Config
from .config import AgentConfig, Config
from .db import Database
from .llm import LLMAdapter
from .router import build_system_prompt, format_messages_for_llm
@ -16,15 +18,70 @@ log = logging.getLogger(__name__)
MAX_TOOL_ITERATIONS = 5
_IMAGE_MIME = {
".png": "image/png",
".jpg": "image/jpeg",
".jpeg": "image/jpeg",
".gif": "image/gif",
".webp": "image/webp",
".bmp": "image/bmp",
}
def _build_file_content_parts(files: list[str]) -> list[dict]:
"""Encode file attachments as content parts for the LLM message.
Images → base64 image_url parts; text files → inline text parts.
"""
parts: list[dict] = []
for file_path in files:
p = Path(file_path).resolve()
if not p.exists():
parts.append({"type": "text", "text": f"[File not found: {file_path}]"})
continue
suffix = p.suffix.lower()
if suffix in _IMAGE_MIME:
try:
data = base64.b64encode(p.read_bytes()).decode("utf-8")
mime = _IMAGE_MIME[suffix]
parts.append({
"type": "image_url",
"image_url": {"url": f"data:{mime};base64,{data}"},
})
except Exception as e:
parts.append({"type": "text", "text": f"[Error reading image {p.name}: {e}]"})
else:
try:
text = p.read_text(encoding="utf-8", errors="replace")
if len(text) > 10000:
text = text[:10000] + "\n... (truncated)"
parts.append({"type": "text", "text": f"[File: {p.name}]\n{text}"})
except Exception as e:
parts.append({"type": "text", "text": f"[Error reading {p.name}: {e}]"})
return parts
class Agent:
def __init__(self, config: Config, db: Database, llm: LLMAdapter):
def __init__(
self,
config: Config,
db: Database,
llm: LLMAdapter,
agent_config: AgentConfig | None = None,
):
self.config = config
self.db = db
self.llm = llm
self.agent_config = agent_config or AgentConfig()
self.conv_id: str | None = None
self._memory = None # set by app after memory system init
self._tools = None # set by app after tool system init
self._skills_registry = None # set by app after skills init
@property
def name(self) -> str:
return self.agent_config.name
def set_memory(self, memory):
self._memory = memory
@ -32,6 +89,9 @@ class Agent:
def set_tools(self, tools):
self._tools = tools
def set_skills_registry(self, registry):
self._skills_registry = registry
def ensure_conversation(self) -> str:
if not self.conv_id:
self.conv_id = uuid.uuid4().hex[:12]
@ -55,26 +115,58 @@ class Agent:
if self._memory:
memory_context = self._memory.get_context(user_input)
# Apply tool whitelist from agent config
tool_filter = self.agent_config.tools
tools_schema = []
tools_description = ""
if self._tools:
tools_schema = self._tools.get_tools_schema()
tools_description = self._tools.get_tools_description()
tools_schema = self._tools.get_tools_schema(filter_names=tool_filter)
tools_description = self._tools.get_tools_description(filter_names=tool_filter)
skills_context = ""
if self._skills_registry:
skills_context = self._skills_registry.get_prompt_section(self.name)
# Use agent-specific personality file if configured
identity_dir = self.config.identity_dir
personality_file = self.agent_config.personality_file
if personality_file:
pf = Path(personality_file)
if pf.exists():
identity_dir = pf.parent
system_prompt = build_system_prompt(
identity_dir=self.config.identity_dir,
identity_dir=identity_dir,
memory_context=memory_context,
tools_description=tools_description,
skills_context=skills_context,
)
# Load conversation history
history = self.db.get_messages(conv_id, limit=self.config.memory.max_context_messages)
messages = format_messages_for_llm(system_prompt, history, self.config.memory.max_context_messages)
messages = format_messages_for_llm(
system_prompt, history, self.config.memory.max_context_messages
)
# If files are attached, replace the last user message with multipart content
if files:
file_parts = _build_file_content_parts(files)
if file_parts:
# Find the last user message and convert to multipart
for i in range(len(messages) - 1, -1, -1):
if messages[i]["role"] == "user":
text_content = messages[i]["content"]
messages[i]["content"] = [
{"type": "text", "text": text_content},
*file_parts,
]
break
# Agent loop: LLM call → tool execution → repeat
seen_tool_calls: set[str] = set() # track (name, args_json) to prevent duplicates
for iteration in range(MAX_TOOL_ITERATIONS):
for _iteration in range(MAX_TOOL_ITERATIONS):
full_response = ""
tool_calls = []
@ -88,7 +180,9 @@ class Agent:
# If no tool calls, we're done
if not tool_calls:
if full_response:
self.db.add_message(conv_id, "assistant", full_response, model=self.llm.current_model)
self.db.add_message(
conv_id, "assistant", full_response, model=self.llm.current_model
)
break
# Filter out duplicate tool calls
@ -104,21 +198,41 @@ class Agent:
if not unique_tool_calls:
# All tool calls were duplicates — force the model to respond
if full_response:
self.db.add_message(conv_id, "assistant", full_response, model=self.llm.current_model)
self.db.add_message(
conv_id, "assistant", full_response, model=self.llm.current_model
)
else:
yield "(I already have the information needed to answer.)"
break
# Store assistant message with tool calls
self.db.add_message(
conv_id, "assistant", full_response,
conv_id,
"assistant",
full_response,
tool_calls=[{"name": tc["name"], "input": tc["input"]} for tc in unique_tool_calls],
model=self.llm.current_model,
)
# Execute tools
if self._tools:
messages.append({"role": "assistant", "content": full_response or "I'll use some tools to help with that."})
# Build OpenAI-format assistant message with tool_calls
openai_tool_calls = [
{
"id": tc.get("id", f"call_{tc['name']}_{i}"),
"type": "function",
"function": {
"name": tc["name"],
"arguments": json.dumps(tc.get("input", {})),
},
}
for i, tc in enumerate(unique_tool_calls)
]
messages.append({
"role": "assistant",
"content": full_response or None,
"tool_calls": openai_tool_calls,
})
for tc in unique_tool_calls:
yield f"\n\n**Using tool: {tc['name']}**\n"
@ -129,11 +243,17 @@ class Agent:
yield f"```\n{result[:2000]}\n```\n\n"
self.db.add_message(conv_id, "tool", result, tool_result=tc["name"])
messages.append({"role": "user", "content": f'[Tool "{tc["name"]}" result]\n{result}'})
messages.append({
"role": "tool",
"tool_call_id": tc.get("id", f"call_{tc['name']}"),
"content": result,
})
else:
# No tool system configured - just mention tool was requested
if full_response:
self.db.add_message(conv_id, "assistant", full_response, model=self.llm.current_model)
self.db.add_message(
conv_id, "assistant", full_response, model=self.llm.current_model
)
for tc in unique_tool_calls:
yield f"\n(Tool requested: {tc['name']} - tool system not yet initialized)\n"
break

49
cheddahbot/agent_registry.py 100644
View File

@ -0,0 +1,49 @@
"""Multi-agent registry.
Holds multiple Agent instances keyed by name. The first registered
agent is the default (used by scheduler and as fallback).
"""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from .agent import Agent
log = logging.getLogger(__name__)
class AgentRegistry:
"""Registry of named Agent instances."""
def __init__(self):
self._agents: dict[str, Agent] = {}
self._default_name: str | None = None
def register(self, name: str, agent: Agent, is_default: bool = False):
"""Register an agent by name."""
self._agents[name] = agent
if is_default or self._default_name is None:
self._default_name = name
log.info("Registered agent: %s (default=%s)", name, name == self._default_name)
def get(self, name: str) -> Agent | None:
return self._agents.get(name)
def list_agents(self) -> list[str]:
return list(self._agents.keys())
@property
def default(self) -> Agent | None:
if self._default_name:
return self._agents.get(self._default_name)
return None
@property
def default_name(self) -> str:
return self._default_name or "default"
def __len__(self) -> int:
return len(self._agents)

25
cheddahbot/clickup.py
View File

@ -43,10 +43,9 @@ class ClickUpTask:
options = cf.get("type_config", {}).get("options", [])
order_index = cf_value if isinstance(cf_value, int) else None
for opt in options:
if order_index is not None and opt.get("orderindex") == order_index:
cf_value = opt.get("name", cf_value)
break
elif opt.get("id") == cf_value:
if (
order_index is not None and opt.get("orderindex") == order_index
) or opt.get("id") == cf_value:
cf_value = opt.get("name", cf_value)
break
@ -72,7 +71,9 @@ class ClickUpTask:
class ClickUpClient:
"""Thin wrapper around the ClickUp REST API v2."""
def __init__(self, api_token: str, workspace_id: str = "", task_type_field_name: str = "Task Type"):
def __init__(
self, api_token: str, workspace_id: str = "", task_type_field_name: str = "Task Type"
):
self._token = api_token
self.workspace_id = workspace_id
self._task_type_field_name = task_type_field_name
@ -110,7 +111,9 @@ class ClickUpClient:
tasks_data = resp.json().get("tasks", [])
return [ClickUpTask.from_api(t, self._task_type_field_name) for t in tasks_data]
def get_tasks_from_space(self, space_id: str, statuses: list[str] | None = None) -> list[ClickUpTask]:
def get_tasks_from_space(
self, space_id: str, statuses: list[str] | None = None
) -> list[ClickUpTask]:
"""Traverse all folders and lists in a space to collect tasks."""
all_tasks: list[ClickUpTask] = []
list_ids = set()
@ -142,7 +145,9 @@ class ClickUpClient:
except httpx.HTTPStatusError as e:
log.warning("Failed to fetch tasks from list %s: %s", list_id, e)
log.info("Found %d tasks across %d lists in space %s", len(all_tasks), len(list_ids), space_id)
log.info(
"Found %d tasks across %d lists in space %s", len(all_tasks), len(list_ids), space_id
)
return all_tasks
# ── Write (with retry) ──
@ -172,10 +177,12 @@ class ClickUpClient:
def update_task_status(self, task_id: str, status: str) -> bool:
"""Update a task's status."""
try:
def _call():
resp = self._client.put(f"/task/{task_id}", json={"status": status})
resp.raise_for_status()
return resp
self._retry(_call)
log.info("Updated task %s status to '%s'", task_id, status)
return True
@ -186,6 +193,7 @@ class ClickUpClient:
def add_comment(self, task_id: str, text: str) -> bool:
"""Add a comment to a task."""
try:
def _call():
resp = self._client.post(
f"/task/{task_id}/comment",
@ -193,6 +201,7 @@ class ClickUpClient:
)
resp.raise_for_status()
return resp
self._retry(_call)
log.info("Added comment to task %s", task_id)
return True
@ -212,6 +221,7 @@ class ClickUpClient:
log.warning("Attachment file not found: %s", fp)
return False
try:
def _call():
with open(fp, "rb") as f:
resp = httpx.post(
@ -222,6 +232,7 @@ class ClickUpClient:
)
resp.raise_for_status()
return resp
self._retry(_call)
log.info("Uploaded attachment %s to task %s", fp.name, task_id)
return True

32
cheddahbot/config.py
View File

@ -29,7 +29,9 @@ class SchedulerConfig:
@dataclass
class ShellConfig:
blocked_commands: list[str] = field(default_factory=lambda: ["rm -rf /", "format", ":(){:|:&};:"])
blocked_commands: list[str] = field(
default_factory=lambda: ["rm -rf /", "format", ":(){:|:&};:"]
)
require_approval: bool = False
@ -64,6 +66,19 @@ class EmailConfig:
enabled: bool = False
@dataclass
class AgentConfig:
"""Per-agent configuration for multi-agent support."""
name: str = "default"
display_name: str = "CheddahBot"
personality_file: str = "" # path to SOUL-like .md file, empty = default
model: str = "" # model override, empty = use global chat_model
tools: list[str] | None = None # tool name whitelist, None = all
skills: list[str] | None = None # skill name filter, None = auto
memory_scope: str = "" # memory namespace, empty = shared
@dataclass
class Config:
chat_model: str = "openai/gpt-4o-mini"
@ -79,6 +94,7 @@ class Config:
clickup: ClickUpConfig = field(default_factory=ClickUpConfig)
press_advantage: PressAdvantageConfig = field(default_factory=PressAdvantageConfig)
email: EmailConfig = field(default_factory=EmailConfig)
agents: list[AgentConfig] = field(default_factory=lambda: [AgentConfig()])
# Derived paths
root_dir: Path = field(default_factory=lambda: ROOT_DIR)
@ -126,6 +142,20 @@ def load_config() -> Config:
if hasattr(cfg.email, k):
setattr(cfg.email, k, v)
# Multi-agent configs
if "agents" in data and isinstance(data["agents"], list):
cfg.agents = []
for agent_data in data["agents"]:
if isinstance(agent_data, dict):
ac = AgentConfig()
for k, v in agent_data.items():
if hasattr(ac, k):
setattr(ac, k, v)
cfg.agents.append(ac)
# Ensure at least one agent
if not cfg.agents:
cfg.agents = [AgentConfig()]
# Env var overrides (CHEDDAH_ prefix)
cfg.openrouter_api_key = os.getenv("OPENROUTER_API_KEY", "")
if cm := os.getenv("CHEDDAH_CHAT_MODEL"):

29
cheddahbot/db.py
View File

@ -5,7 +5,7 @@ from __future__ import annotations
import json
import sqlite3
import threading
from datetime import datetime, timezone
from datetime import UTC, datetime
from pathlib import Path
@ -105,7 +105,8 @@ class Database:
) -> int:
now = _now()
cur = self._conn.execute(
"""INSERT INTO messages (conv_id, role, content, tool_calls, tool_result, model, created_at)
"""INSERT INTO messages
(conv_id, role, content, tool_calls, tool_result, model, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?)""",
(
conv_id,
@ -117,9 +118,7 @@ class Database:
now,
),
)
self._conn.execute(
"UPDATE conversations SET updated_at = ? WHERE id = ?", (now, conv_id)
)
self._conn.execute("UPDATE conversations SET updated_at = ? WHERE id = ?", (now, conv_id))
self._conn.commit()
return cur.lastrowid
@ -148,9 +147,7 @@ class Database:
if not message_ids:
return
placeholders = ",".join("?" for _ in message_ids)
self._conn.execute(
f"DELETE FROM messages WHERE id IN ({placeholders})", message_ids
)
self._conn.execute(f"DELETE FROM messages WHERE id IN ({placeholders})", message_ids)
self._conn.commit()
# -- Scheduled Tasks --
@ -167,7 +164,8 @@ class Database:
def get_due_tasks(self) -> list[dict]:
now = _now()
rows = self._conn.execute(
"SELECT * FROM scheduled_tasks WHERE enabled = 1 AND (next_run IS NULL OR next_run <= ?)",
"SELECT * FROM scheduled_tasks"
" WHERE enabled = 1 AND (next_run IS NULL OR next_run <= ?)",
(now,),
).fetchall()
return [dict(r) for r in rows]
@ -180,15 +178,15 @@ class Database:
def disable_task(self, task_id: int):
"""Disable a scheduled task (e.g. after a one-time task has run)."""
self._conn.execute(
"UPDATE scheduled_tasks SET enabled = 0 WHERE id = ?", (task_id,)
)
self._conn.execute("UPDATE scheduled_tasks SET enabled = 0 WHERE id = ?", (task_id,))
self._conn.commit()
def log_task_run(self, task_id: int, result: str | None = None, error: str | None = None):
now = _now()
self._conn.execute(
"INSERT INTO task_run_logs (task_id, started_at, finished_at, result, error) VALUES (?, ?, ?, ?, ?)",
"INSERT INTO task_run_logs"
" (task_id, started_at, finished_at, result, error)"
" VALUES (?, ?, ?, ?, ?)",
(task_id, now, now, result, error),
)
self._conn.commit()
@ -231,11 +229,12 @@ class Database:
def get_notifications_after(self, after_id: int = 0, limit: int = 50) -> list[dict]:
"""Get notifications with id > after_id."""
rows = self._conn.execute(
"SELECT id, message, category, created_at FROM notifications WHERE id > ? ORDER BY id ASC LIMIT ?",
"SELECT id, message, category, created_at FROM notifications"
" WHERE id > ? ORDER BY id ASC LIMIT ?",
(after_id, limit),
).fetchall()
return [dict(r) for r in rows]
def _now() -> str:
return datetime.now(timezone.utc).isoformat()
return datetime.now(UTC).isoformat()

71
cheddahbot/llm.py
View File

@ -19,8 +19,8 @@ import os
import shutil
import subprocess
import sys
from collections.abc import Generator
from dataclasses import dataclass
from typing import Generator
import httpx
@ -96,22 +96,28 @@ class LLMAdapter:
model_id = CLAUDE_OPENROUTER_MAP[model_id]
provider = "openrouter"
else:
yield {"type": "text", "content": (
yield {
"type": "text",
"content": (
"To chat with Claude models, you need an OpenRouter API key "
"(set OPENROUTER_API_KEY in .env). Alternatively, select a local "
"model from Ollama or LM Studio."
)}
),
}
return
# Check if provider is available
if provider == "openrouter" and not self.openrouter_key:
yield {"type": "text", "content": (
yield {
"type": "text",
"content": (
"No API key configured. To use cloud models:\n"
"1. Get an OpenRouter API key at https://openrouter.ai/keys\n"
"2. Set OPENROUTER_API_KEY in your .env file\n\n"
"Or install Ollama (free, local) and pull a model:\n"
" ollama pull llama3.2"
)}
),
}
return
base_url, api_key = self._resolve_endpoint(provider)
@ -138,14 +144,21 @@ class LLMAdapter:
"""
claude_bin = shutil.which("claude")
if not claude_bin:
return "Error: `claude` CLI not found in PATH. Install Claude Code: npm install -g @anthropic-ai/claude-code"
return (
"Error: `claude` CLI not found in PATH. "
"Install Claude Code: npm install -g @anthropic-ai/claude-code"
)
# Pipe prompt through stdin to avoid Windows 8191-char command-line limit.
cmd = [
claude_bin, "-p",
"--output-format", "json",
"--tools", tools,
"--allowedTools", tools,
claude_bin,
"-p",
"--output-format",
"json",
"--tools",
tools,
"--allowedTools",
tools,
]
if model:
cmd.extend(["--model", model])
@ -170,7 +183,10 @@ class LLMAdapter:
env=env,
)
except FileNotFoundError:
return "Error: `claude` CLI not found. Install Claude Code: npm install -g @anthropic-ai/claude-code"
return (
"Error: `claude` CLI not found. "
"Install Claude Code: npm install -g @anthropic-ai/claude-code"
)
try:
stdout, stderr = proc.communicate(input=prompt, timeout=300)
@ -234,7 +250,9 @@ class LLMAdapter:
if idx not in tool_calls_accum:
tool_calls_accum[idx] = {
"id": tc.id or "",
"name": tc.function.name if tc.function and tc.function.name else "",
"name": tc.function.name
if tc.function and tc.function.name
else "",
"arguments": "",
}
if tc.function and tc.function.arguments:
@ -276,7 +294,7 @@ class LLMAdapter:
# ── Helpers ──
def _resolve_endpoint(self, provider: str) -> tuple[str, str]:
if provider == "openrouter":
if provider == "openrouter": # noqa: SIM116
return "https://openrouter.ai/api/v1", self.openrouter_key or "sk-placeholder"
elif provider == "ollama":
return f"{self.ollama_url}/v1", "ollama"
@ -295,6 +313,7 @@ class LLMAdapter:
def _get_openai(self):
if self._openai_mod is None:
import openai
self._openai_mod = openai
return self._openai_mod
@ -307,11 +326,13 @@ class LLMAdapter:
r = httpx.get(f"{self.ollama_url}/api/tags", timeout=3)
if r.status_code == 200:
for m in r.json().get("models", []):
models.append(ModelInfo(
models.append(
ModelInfo(
id=f"local/ollama/{m['name']}",
name=f"[Ollama] {m['name']}",
provider="ollama",
))
)
)
except Exception:
pass
# LM Studio
@ -319,11 +340,13 @@ class LLMAdapter:
r = httpx.get(f"{self.lmstudio_url}/v1/models", timeout=3)
if r.status_code == 200:
for m in r.json().get("data", []):
models.append(ModelInfo(
models.append(
ModelInfo(
id=f"local/lmstudio/{m['id']}",
name=f"[LM Studio] {m['id']}",
provider="lmstudio",
))
)
)
except Exception:
pass
return models
@ -333,14 +356,19 @@ class LLMAdapter:
models = []
if self.openrouter_key:
models.extend([
models.extend(
[
# Anthropic (via OpenRouter — system prompts work correctly)
ModelInfo("anthropic/claude-sonnet-4.5", "Claude Sonnet 4.5", "openrouter"),
ModelInfo("anthropic/claude-opus-4.6", "Claude Opus 4.6", "openrouter"),
# Google
ModelInfo("google/gemini-3-flash-preview", "Gemini 3 Flash Preview", "openrouter"),
ModelInfo(
"google/gemini-3-flash-preview", "Gemini 3 Flash Preview", "openrouter"
),
ModelInfo("google/gemini-2.5-flash", "Gemini 2.5 Flash", "openrouter"),
ModelInfo("google/gemini-2.5-flash-lite", "Gemini 2.5 Flash Lite", "openrouter"),
ModelInfo(
"google/gemini-2.5-flash-lite", "Gemini 2.5 Flash Lite", "openrouter"
),
# OpenAI
ModelInfo("openai/gpt-5-nano", "GPT-5 Nano", "openrouter"),
ModelInfo("openai/gpt-4o-mini", "GPT-4o Mini", "openrouter"),
@ -349,7 +377,8 @@ class LLMAdapter:
ModelInfo("x-ai/grok-4.1-fast", "Grok 4.1 Fast", "openrouter"),
ModelInfo("moonshotai/kimi-k2.5", "Kimi K2.5", "openrouter"),
ModelInfo("minimax/minimax-m2.5", "MiniMax M2.5", "openrouter"),
])
]
)
models.extend(self.discover_local_models())
return models

50
cheddahbot/media.py
View File

@ -13,6 +13,7 @@ log = logging.getLogger(__name__)
# ── Speech-to-Text ──
def transcribe_audio(audio_path: str | Path) -> str:
"""Transcribe audio to text. Tries OpenAI Whisper API, falls back to local whisper."""
audio_path = Path(audio_path)
@ -38,14 +39,17 @@ def transcribe_audio(audio_path: str | Path) -> str:
def _transcribe_local(audio_path: Path) -> str:
import whisper
model = whisper.load_model("base")
result = model.transcribe(str(audio_path))
return result.get("text", "").strip()
def _transcribe_openai_api(audio_path: Path) -> str:
import openai
import os
import openai
key = os.getenv("OPENAI_API_KEY") or os.getenv("OPENROUTER_API_KEY")
if not key:
raise ValueError("No API key for Whisper")
@ -57,18 +61,20 @@ def _transcribe_openai_api(audio_path: Path) -> str:
# ── Text-to-Speech ──
def text_to_speech(text: str, output_path: str | Path | None = None, voice: str = "en-US-AriaNeural") -> Path:
def text_to_speech(
text: str, output_path: str | Path | None = None, voice: str = "en-US-AriaNeural"
) -> Path:
"""Convert text to speech using edge-tts (free, no API key)."""
if output_path is None:
output_path = Path(tempfile.mktemp(suffix=".mp3"))
else:
output_path = Path(output_path)
output_path = Path(tempfile.mktemp(suffix=".mp3")) if output_path is None else Path(output_path)
try:
import edge_tts
async def _generate():
communicate = edge_tts.Communicate(text, voice)
await communicate.save(str(output_path))
asyncio.run(_generate())
return output_path
except ImportError:
@ -80,6 +86,7 @@ def text_to_speech(text: str, output_path: str | Path | None = None, voice: str
# ── Video Frame Extraction ──
def extract_video_frames(video_path: str | Path, max_frames: int = 5) -> list[Path]:
"""Extract key frames from a video using ffmpeg."""
video_path = Path(video_path)
@ -91,18 +98,37 @@ def extract_video_frames(video_path: str | Path, max_frames: int = 5) -> list[Pa
try:
# Get video duration
result = subprocess.run(
["ffprobe", "-v", "error", "-show_entries", "format=duration",
"-of", "default=noprint_wrappers=1:nokey=1", str(video_path)],
capture_output=True, text=True, timeout=10,
[
"ffprobe",
"-v",
"error",
"-show_entries",
"format=duration",
"-of",
"default=noprint_wrappers=1:nokey=1",
str(video_path),
],
capture_output=True,
text=True,
timeout=10,
)
duration = float(result.stdout.strip()) if result.stdout.strip() else 10.0
interval = max(duration / (max_frames + 1), 1.0)
# Extract frames
subprocess.run(
["ffmpeg", "-i", str(video_path), "-vf", f"fps=1/{interval}",
"-frames:v", str(max_frames), str(output_dir / "frame_%03d.jpg")],
capture_output=True, timeout=30,
[
"ffmpeg",
"-i",
str(video_path),
"-vf",
f"fps=1/{interval}",
"-frames:v",
str(max_frames),
str(output_dir / "frame_%03d.jpg"),
],
capture_output=True,
timeout=30,
)
frames = sorted(output_dir.glob("frame_*.jpg"))

78
cheddahbot/memory.py
View File

@ -12,8 +12,7 @@ from __future__ import annotations
import logging
import sqlite3
import threading
from datetime import datetime, timezone
from pathlib import Path
from datetime import UTC, datetime
import numpy as np
@ -24,13 +23,25 @@ log = logging.getLogger(__name__)
class MemorySystem:
def __init__(self, config: Config, db: Database):
def __init__(self, config: Config, db: Database, scope: str = ""):
self.config = config
self.db = db
self.scope = scope
# Scoped agents get their own subdirectory; shared memory stays in root
if scope:
self.memory_dir = config.memory_dir / scope
self.memory_dir.mkdir(parents=True, exist_ok=True)
else:
self.memory_dir = config.memory_dir
# Shared memory dir (for cross-scope search)
self._shared_memory_dir = config.memory_dir
self._embedder = None
self._embed_lock = threading.Lock()
self._embed_db_path = self.memory_dir / "embeddings.db"
self._embed_local = threading.local()
self._init_embed_db()
# ── Public API ──
@ -61,7 +72,7 @@ class MemorySystem:
def remember(self, text: str):
"""Save a fact/instruction to long-term memory."""
memory_path = self.memory_dir / "MEMORY.md"
timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M")
timestamp = datetime.now(UTC).strftime("%Y-%m-%d %H:%M")
entry = f"\n- [{timestamp}] {text}\n"
if memory_path.exists():
@ -76,9 +87,9 @@ class MemorySystem:
def log_daily(self, text: str):
"""Append an entry to today's daily log."""
today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
today = datetime.now(UTC).strftime("%Y-%m-%d")
log_path = self.memory_dir / f"{today}.md"
timestamp = datetime.now(timezone.utc).strftime("%H:%M")
timestamp = datetime.now(UTC).strftime("%H:%M")
if log_path.exists():
content = log_path.read_text(encoding="utf-8")
@ -121,7 +132,9 @@ class MemorySystem:
if not summary_parts:
return
summary = f"Conversation summary ({len(to_summarize)} messages):\n" + "\n".join(summary_parts[:20])
summary = f"Conversation summary ({len(to_summarize)} messages):\n" + "\n".join(
summary_parts[:20]
)
self.log_daily(summary)
# Delete the flushed messages from DB so they don't get re-flushed
@ -153,7 +166,7 @@ class MemorySystem:
return ""
def _read_daily_log(self) -> str:
today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
today = datetime.now(UTC).strftime("%Y-%m-%d")
path = self.memory_dir / f"{today}.md"
if path.exists():
content = path.read_text(encoding="utf-8")
@ -162,17 +175,23 @@ class MemorySystem:
# ── Private: Embedding system ──
@property
def _embed_conn(self) -> sqlite3.Connection:
"""Thread-local SQLite connection for embeddings DB (matches db.py pattern)."""
if not hasattr(self._embed_local, "conn"):
self._embed_local.conn = sqlite3.connect(str(self._embed_db_path))
self._embed_local.conn.execute("PRAGMA journal_mode=WAL")
return self._embed_local.conn
def _init_embed_db(self):
conn = sqlite3.connect(str(self._embed_db_path))
conn.execute("""
self._embed_conn.execute("""
CREATE TABLE IF NOT EXISTS embeddings (
id TEXT PRIMARY KEY,
text TEXT NOT NULL,
vector BLOB NOT NULL
)
""")
conn.commit()
conn.close()
self._embed_conn.commit()
def _get_embedder(self):
if self._embedder is not None:
@ -182,6 +201,7 @@ class MemorySystem:
return self._embedder
try:
from sentence_transformers import SentenceTransformer
model_name = self.config.memory.embedding_model
log.info("Loading embedding model: %s", model_name)
self._embedder = SentenceTransformer(model_name)
@ -198,18 +218,14 @@ class MemorySystem:
if embedder is None:
return
vec = embedder.encode([text])[0]
conn = sqlite3.connect(str(self._embed_db_path))
conn.execute(
self._embed_conn.execute(
"INSERT OR REPLACE INTO embeddings (id, text, vector) VALUES (?, ?, ?)",
(doc_id, text, vec.tobytes()),
)
conn.commit()
conn.close()
self._embed_conn.commit()
def _vector_search(self, query_vec: np.ndarray, top_k: int) -> list[dict]:
conn = sqlite3.connect(str(self._embed_db_path))
rows = conn.execute("SELECT id, text, vector FROM embeddings").fetchall()
conn.close()
rows = self._embed_conn.execute("SELECT id, text, vector FROM embeddings").fetchall()
if not rows:
return []
@ -217,23 +233,33 @@ class MemorySystem:
scored = []
for doc_id, text, vec_bytes in rows:
vec = np.frombuffer(vec_bytes, dtype=np.float32)
sim = float(np.dot(query_vec, vec) / (np.linalg.norm(query_vec) * np.linalg.norm(vec) + 1e-8))
sim = float(
np.dot(query_vec, vec) / (np.linalg.norm(query_vec) * np.linalg.norm(vec) + 1e-8)
)
scored.append({"id": doc_id, "text": text, "score": sim})
scored.sort(key=lambda x: x["score"], reverse=True)
return scored[:top_k]
def _clear_embeddings(self):
conn = sqlite3.connect(str(self._embed_db_path))
conn.execute("DELETE FROM embeddings")
conn.commit()
conn.close()
self._embed_conn.execute("DELETE FROM embeddings")
self._embed_conn.commit()
def _fallback_search(self, query: str, top_k: int) -> list[dict]:
"""Simple keyword search when embeddings are unavailable."""
"""Simple keyword search when embeddings are unavailable.
Searches both scoped and shared memory directories.
"""
results = []
query_lower = query.lower()
for path in self.memory_dir.glob("*.md"):
# Collect directories to search (avoid duplicates)
search_dirs = [self.memory_dir]
if self.scope and self._shared_memory_dir != self.memory_dir:
search_dirs.append(self._shared_memory_dir)
for search_dir in search_dirs:
for path in search_dir.glob("*.md"):
try:
content = path.read_text(encoding="utf-8")
except Exception:

3
cheddahbot/notifications.py
View File

@ -9,7 +9,8 @@ from __future__ import annotations
import logging
import threading
from typing import Callable, TYPE_CHECKING
from collections.abc import Callable
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from .db import Database

1
cheddahbot/providers/__init__.py
View File

@ -1 +0,0 @@
# Reserved for future custom providers

9
cheddahbot/router.py
View File

@ -9,8 +9,9 @@ def build_system_prompt(
identity_dir: Path,
memory_context: str = "",
tools_description: str = "",
skills_context: str = "",
) -> str:
"""Build the system prompt from identity files + memory + tools."""
"""Build the system prompt from identity files + memory + skills + tools."""
parts = []
# 1. Identity: SOUL.md
@ -27,7 +28,11 @@ def build_system_prompt(
if memory_context:
parts.append(f"# Relevant Memory\n{memory_context}")
# 4. Available tools
# 4. Skills context (injected by skill registry)
if skills_context:
parts.append(skills_context)
# 5. Available tools
if tools_description:
parts.append(f"# Available Tools\n{tools_description}")

50
cheddahbot/scheduler.py
View File

@ -6,7 +6,7 @@ import json
import logging
import re
import threading
from datetime import datetime, timezone
from datetime import UTC, datetime
from typing import TYPE_CHECKING
from croniter import croniter
@ -31,8 +31,13 @@ def _extract_docx_paths(result: str) -> list[str]:
class Scheduler:
def __init__(self, config: Config, db: Database, agent: Agent,
notification_bus: NotificationBus | None = None):
def __init__(
self,
config: Config,
db: Database,
agent: Agent,
notification_bus: NotificationBus | None = None,
):
self.config = config
self.db = db
self.agent = agent
@ -48,20 +53,28 @@ class Scheduler:
self._thread = threading.Thread(target=self._poll_loop, daemon=True, name="scheduler")
self._thread.start()
self._heartbeat_thread = threading.Thread(target=self._heartbeat_loop, daemon=True, name="heartbeat")
self._heartbeat_thread = threading.Thread(
target=self._heartbeat_loop, daemon=True, name="heartbeat"
)
self._heartbeat_thread.start()
# Start ClickUp polling if configured
if self.config.clickup.enabled:
self._clickup_thread = threading.Thread(target=self._clickup_loop, daemon=True, name="clickup")
self._clickup_thread = threading.Thread(
target=self._clickup_loop, daemon=True, name="clickup"
)
self._clickup_thread.start()
log.info("ClickUp polling started (interval=%dm)", self.config.clickup.poll_interval_minutes)
log.info(
"ClickUp polling started (interval=%dm)", self.config.clickup.poll_interval_minutes
)
else:
log.info("ClickUp integration disabled (no API token)")
log.info("Scheduler started (poll=%ds, heartbeat=%dm)",
log.info(
"Scheduler started (poll=%ds, heartbeat=%dm)",
self.config.scheduler.poll_interval_seconds,
self.config.scheduler.heartbeat_interval_minutes)
self.config.scheduler.heartbeat_interval_minutes,
)
def stop(self):
self._stop_event.set()
@ -100,7 +113,7 @@ class Scheduler:
self.db.disable_task(task["id"])
else:
# Cron schedule - calculate next run
now = datetime.now(timezone.utc)
now = datetime.now(UTC)
cron = croniter(schedule, now)
next_run = cron.get_next(datetime)
self.db.update_task_next_run(task["id"], next_run.isoformat())
@ -147,6 +160,7 @@ class Scheduler:
"""Lazy-init the ClickUp API client."""
if self._clickup_client is None:
from .clickup import ClickUpClient
self._clickup_client = ClickUpClient(
api_token=self.config.clickup.api_token,
workspace_id=self.config.clickup.workspace_id,
@ -216,9 +230,8 @@ class Scheduler:
def _process_clickup_task(self, task, active_ids: set[str]):
"""Discover a new ClickUp task, map to skill, decide action."""
from .clickup import ClickUpTask
now = datetime.now(timezone.utc).isoformat()
now = datetime.now(UTC).isoformat()
skill_map = self.config.clickup.skill_map
# Build state object
@ -270,8 +283,8 @@ class Scheduler:
self._notify(
f"New ClickUp task needs your approval.\n"
f"Task: **{task.name}** → Skill: `{tool_name}`\n"
f"Use `clickup_approve_task(\"{task.id}\")` to approve or "
f"`clickup_decline_task(\"{task.id}\")` to decline."
f'Use `clickup_approve_task("{task.id}")` to approve or '
f'`clickup_decline_task("{task.id}")` to decline.'
)
log.info("ClickUp task awaiting approval: %s → %s", task.name, tool_name)
@ -296,7 +309,7 @@ class Scheduler:
task_id = state["clickup_task_id"]
task_name = state["clickup_task_name"]
skill_name = state["skill_name"]
now = datetime.now(timezone.utc).isoformat()
now = datetime.now(UTC).isoformat()
log.info("Executing ClickUp task: %s → %s", task_name, skill_name)
@ -314,7 +327,7 @@ class Scheduler:
args = self._build_tool_args(state)
# Execute the skill via the tool registry
if hasattr(self.agent, '_tools') and self.agent._tools:
if hasattr(self.agent, "_tools") and self.agent._tools:
result = self.agent._tools.execute(skill_name, args)
else:
result = self.agent.execute_task(
@ -334,7 +347,7 @@ class Scheduler:
# Success
state["state"] = "completed"
state["completed_at"] = datetime.now(timezone.utc).isoformat()
state["completed_at"] = datetime.now(UTC).isoformat()
self.db.kv_set(kv_key, json.dumps(state))
# Update ClickUp
@ -357,13 +370,12 @@ class Scheduler:
# Failure
state["state"] = "failed"
state["error"] = str(e)
state["completed_at"] = datetime.now(timezone.utc).isoformat()
state["completed_at"] = datetime.now(UTC).isoformat()
self.db.kv_set(kv_key, json.dumps(state))
# Comment the error on ClickUp
client.add_comment(
task_id,
f"❌ CheddahBot failed to complete this task.\n\nError: {str(e)[:2000]}"
task_id, f"❌ CheddahBot failed to complete this task.\n\nError: {str(e)[:2000]}"
)
self._notify(

132
cheddahbot/skills.py 100644
View File

@ -0,0 +1,132 @@
"""Markdown skill registry.
Skills are .md files in the skills/ directory with YAML frontmatter:
---
name: press-release-writer
description: Write professional press releases
tools: [write_press_releases, submit_press_release]
agents: [writer, default]
---
# Press Release Workflow
...
Files without frontmatter (like companies.md, headlines.md) are data
files and get skipped.
"""
from __future__ import annotations
import logging
from dataclasses import dataclass, field
from pathlib import Path
import yaml
log = logging.getLogger(__name__)
@dataclass
class SkillDef:
name: str
description: str
content: str
file_path: Path
tools: list[str] = field(default_factory=list)
agents: list[str] = field(default_factory=list)
def _parse_frontmatter(text: str) -> tuple[dict, str]:
"""Split YAML frontmatter from markdown content.
Returns (frontmatter_dict, body) or ({}, full_text) if no frontmatter.
"""
if not text.startswith("---"):
return {}, text
end = text.find("---", 3)
if end == -1:
return {}, text
yaml_block = text[3:end].strip()
body = text[end + 3 :].strip()
try:
meta = yaml.safe_load(yaml_block) or {}
except yaml.YAMLError as e:
log.warning("Failed to parse YAML frontmatter: %s", e)
return {}, text
if not isinstance(meta, dict):
return {}, text
return meta, body
class SkillRegistry:
"""Discovers and holds markdown skill definitions."""
def __init__(self, skills_dir: Path):
self._skills: dict[str, SkillDef] = {}
self._skills_dir = skills_dir
self._discover()
def _discover(self):
"""Load all .md files with valid frontmatter from the skills directory."""
if not self._skills_dir.exists():
log.warning("Skills directory not found: %s", self._skills_dir)
return
for path in sorted(self._skills_dir.glob("*.md")):
try:
text = path.read_text(encoding="utf-8")
except Exception as e:
log.warning("Failed to read skill file %s: %s", path.name, e)
continue
meta, body = _parse_frontmatter(text)
if not meta.get("name"):
# No frontmatter or no name — skip (data file)
continue
skill = SkillDef(
name=meta["name"],
description=meta.get("description", ""),
content=body,
file_path=path,
tools=meta.get("tools", []),
agents=meta.get("agents", []),
)
self._skills[skill.name] = skill
log.info("Loaded skill: %s (%s)", skill.name, path.name)
def get(self, name: str) -> SkillDef | None:
return self._skills.get(name)
def list_skills(self) -> list[SkillDef]:
return list(self._skills.values())
def get_prompt_section(self, agent_name: str = "default") -> str:
"""Build a prompt section with skills relevant to an agent.
If a skill's agents list is empty, it's available to all agents.
Otherwise, only agents listed in the skill's agents list get it.
"""
parts = []
for skill in self._skills.values():
if skill.agents and agent_name not in skill.agents:
continue
parts.append(
f"### Skill: {skill.name}\n"
f"{skill.description}\n"
)
if not parts:
return ""
return "# Available Skills\n" + "\n".join(parts)
def get_body(self, name: str) -> str:
"""Get the body content of a skill (without frontmatter)."""
skill = self._skills.get(name)
return skill.content if skill else ""

63
cheddahbot/skills/__init__.py
View File

@ -1,63 +0,0 @@
"""Skill registry with @skill decorator and loader."""
from __future__ import annotations
import importlib.util
import logging
from pathlib import Path
from typing import Callable
log = logging.getLogger(__name__)
_SKILLS: dict[str, "SkillDef"] = {}
class SkillDef:
def __init__(self, name: str, description: str, func: Callable):
self.name = name
self.description = description
self.func = func
def skill(name: str, description: str):
"""Decorator to register a skill."""
def decorator(func: Callable) -> Callable:
_SKILLS[name] = SkillDef(name, description, func)
return func
return decorator
def load_skill(path: Path):
"""Dynamically load a skill from a .py file."""
spec = importlib.util.spec_from_file_location(path.stem, path)
if spec and spec.loader:
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
log.info("Loaded skill from %s", path)
def discover_skills(skills_dir: Path):
"""Load all .py files from the skills directory."""
if not skills_dir.exists():
return
for path in skills_dir.glob("*.py"):
if path.name.startswith("_"):
continue
try:
load_skill(path)
except Exception as e:
log.warning("Failed to load skill %s: %s", path.name, e)
def list_skills() -> list[SkillDef]:
return list(_SKILLS.values())
def run_skill(name: str, **kwargs) -> str:
if name not in _SKILLS:
return f"Unknown skill: {name}"
try:
result = _SKILLS[name].func(**kwargs)
return str(result) if result is not None else "Done."
except Exception as e:
return f"Skill error: {e}"

33
cheddahbot/tools/__init__.py
View File

@ -4,11 +4,11 @@ from __future__ import annotations
import importlib
import inspect
import json
import logging
import pkgutil
from collections.abc import Callable
from pathlib import Path
from typing import Any, Callable, TYPE_CHECKING
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from ..agent import Agent
@ -72,15 +72,15 @@ def _extract_params(func: Callable) -> dict:
prop: dict[str, Any] = {}
annotation = param.annotation
if annotation == str or annotation == inspect.Parameter.empty:
if annotation is str or annotation is inspect.Parameter.empty:
prop["type"] = "string"
elif annotation == int:
elif annotation is int:
prop["type"] = "integer"
elif annotation == float:
elif annotation is float:
prop["type"] = "number"
elif annotation == bool:
elif annotation is bool:
prop["type"] = "boolean"
elif annotation == list:
elif annotation is list:
prop["type"] = "array"
prop["items"] = {"type": "string"}
else:
@ -100,10 +100,11 @@ def _extract_params(func: Callable) -> dict:
class ToolRegistry:
"""Runtime tool registry with execution and schema generation."""
def __init__(self, config: "Config", db: "Database", agent: "Agent"):
def __init__(self, config: Config, db: Database, agent: Agent):
self.config = config
self.db = db
self.agent = agent
self.agent_registry = None # set after multi-agent setup
self._discover_tools()
def _discover_tools(self):
@ -118,15 +119,20 @@ class ToolRegistry:
except Exception as e:
log.warning("Failed to load tool module %s: %s", module_name, e)
def get_tools_schema(self) -> list[dict]:
"""Get all tools in OpenAI function-calling format."""
return [t.to_openai_schema() for t in _TOOLS.values()]
def get_tools_schema(self, filter_names: list[str] | None = None) -> list[dict]:
"""Get tools in OpenAI function-calling format, optionally filtered."""
tools = _TOOLS.values()
if filter_names is not None:
tools = [t for t in tools if t.name in filter_names]
return [t.to_openai_schema() for t in tools]
def get_tools_description(self) -> str:
"""Human-readable tool list for system prompt."""
def get_tools_description(self, filter_names: list[str] | None = None) -> str:
"""Human-readable tool list for system prompt, optionally filtered."""
lines = []
by_cat: dict[str, list[ToolDef]] = {}
for t in _TOOLS.values():
if filter_names is not None and t.name not in filter_names:
continue
by_cat.setdefault(t.category, []).append(t)
for cat, tools in sorted(by_cat.items()):
@ -151,6 +157,7 @@ class ToolRegistry:
"db": self.db,
"agent": self.agent,
"memory": self.agent._memory,
"agent_registry": self.agent_registry,
}
result = tool_def.func(**args)
return str(result) if result is not None else "Done."

49
cheddahbot/tools/build_skill.py
View File

@ -1,49 +0,0 @@
"""Meta-skill: create multi-step skills at runtime."""
from __future__ import annotations
import textwrap
from pathlib import Path
from . import tool
@tool("build_skill", "Create a new multi-step skill from a description", category="meta")
def build_skill(name: str, description: str, steps: str, ctx: dict = None) -> str:
"""Generate a new skill and save it to the skills directory.
Args:
name: Skill name (snake_case)
description: What the skill does
steps: Python code implementing the skill steps (must use @skill decorator)
"""
if not name.isidentifier():
return f"Invalid skill name: {name}. Must be a valid Python identifier."
if not ctx or not ctx.get("config"):
return "Config context not available."
skills_dir = ctx["config"].skills_dir
skills_dir.mkdir(parents=True, exist_ok=True)
module_code = textwrap.dedent(f'''\
"""Auto-generated skill: {description}"""
from __future__ import annotations
from cheddahbot.skills import skill
{steps}
''')
file_path = skills_dir / f"{name}.py"
if file_path.exists():
return f"Skill '{name}' already exists. Choose a different name."
file_path.write_text(module_code, encoding="utf-8")
# Try to load it
try:
from cheddahbot.skills import load_skill
load_skill(file_path)
return f"Skill '{name}' created at {file_path}"
except Exception as e:
return f"Skill created at {file_path} but failed to load: {e}"

48
cheddahbot/tools/build_tool.py
View File

@ -1,48 +0,0 @@
"""Meta-tool: dynamically create new tools at runtime."""
from __future__ import annotations
import importlib
import textwrap
from pathlib import Path
from . import tool
@tool("build_tool", "Create a new tool from a description. The agent writes Python code with @tool decorator.", category="meta")
def build_tool(name: str, description: str, code: str, ctx: dict = None) -> str:
"""Generate a new tool module and hot-load it.
Args:
name: Tool name (snake_case)
description: What the tool does
code: Full Python code for the tool function (must use @tool decorator)
"""
if not name.isidentifier():
return f"Invalid tool name: {name}. Must be a valid Python identifier."
# Wrap code in a module with the import
module_code = textwrap.dedent(f'''\
"""Auto-generated tool: {description}"""
from __future__ import annotations
from . import tool
{code}
''')
# Write to tools directory
tools_dir = Path(__file__).parent
file_path = tools_dir / f"{name}.py"
if file_path.exists():
return f"Tool module '{name}' already exists. Choose a different name."
file_path.write_text(module_code, encoding="utf-8")
# Hot-import the new module
try:
importlib.import_module(f".{name}", package=__package__)
return f"Tool '{name}' created and loaded successfully at {file_path}"
except Exception as e:
# Clean up on failure
file_path.unlink(missing_ok=True)
return f"Failed to load tool '{name}': {e}"

24
cheddahbot/tools/calendar_tool.py
View File

@ -2,13 +2,13 @@
from __future__ import annotations
from datetime import datetime, timezone
from . import tool
@tool("remember_this", "Save an important fact or instruction to long-term memory", category="memory")
def remember_this(text: str, ctx: dict = None) -> str:
@tool(
"remember_this", "Save an important fact or instruction to long-term memory", category="memory"
)
def remember_this(text: str, ctx: dict | None = None) -> str:
if ctx and ctx.get("memory"):
ctx["memory"].remember(text)
return f"Saved to memory: {text}"
@ -16,7 +16,7 @@ def remember_this(text: str, ctx: dict = None) -> str:
@tool("search_memory", "Search through saved memories", category="memory")
def search_memory(query: str, ctx: dict = None) -> str:
def search_memory(query: str, ctx: dict | None = None) -> str:
if ctx and ctx.get("memory"):
results = ctx["memory"].search(query)
if results:
@ -26,7 +26,7 @@ def search_memory(query: str, ctx: dict = None) -> str:
@tool("log_note", "Add a timestamped note to today's daily log", category="memory")
def log_note(text: str, ctx: dict = None) -> str:
def log_note(text: str, ctx: dict | None = None) -> str:
if ctx and ctx.get("memory"):
ctx["memory"].log_daily(text)
return f"Logged: {text}"
@ -34,7 +34,7 @@ def log_note(text: str, ctx: dict = None) -> str:
@tool("schedule_task", "Schedule a recurring or one-time task", category="scheduling")
def schedule_task(name: str, prompt: str, schedule: str, ctx: dict = None) -> str:
def schedule_task(name: str, prompt: str, schedule: str, ctx: dict | None = None) -> str:
"""Schedule a task. Schedule format: cron expression or 'once:YYYY-MM-DDTHH:MM'."""
if ctx and ctx.get("db"):
task_id = ctx["db"].add_scheduled_task(name, prompt, schedule)
@ -43,11 +43,15 @@ def schedule_task(name: str, prompt: str, schedule: str, ctx: dict = None) -> st
@tool("list_tasks", "List all scheduled tasks", category="scheduling")
def list_tasks(ctx: dict = None) -> str:
def list_tasks(ctx: dict | None = None) -> str:
if ctx and ctx.get("db"):
tasks = ctx["db"]._conn.execute(
tasks = (
ctx["db"]
._conn.execute(
"SELECT id, name, schedule, enabled, next_run FROM scheduled_tasks ORDER BY id"
).fetchall()
)
.fetchall()
)
if not tasks:
return "No scheduled tasks."
lines = []

21
cheddahbot/tools/clickup_tool.py
View File

@ -33,7 +33,7 @@ def _get_clickup_states(db) -> dict[str, dict]:
parts = key.split(":")
if len(parts) == 4 and parts[3] == "state":
task_id = parts[2]
try:
try: # noqa: SIM105
states[task_id] = json.loads(value)
except json.JSONDecodeError:
pass
@ -47,7 +47,7 @@ def _get_clickup_states(db) -> dict[str, dict]:
"and custom fields directly from the ClickUp API.",
category="clickup",
)
def clickup_query_tasks(status: str = "", task_type: str = "", ctx: dict = None) -> str:
def clickup_query_tasks(status: str = "", task_type: str = "", ctx: dict | None = None) -> str:
"""Query ClickUp API for tasks, optionally filtered by status and task type."""
client = _get_clickup_client(ctx)
if not client:
@ -98,7 +98,7 @@ def clickup_query_tasks(status: str = "", task_type: str = "", ctx: dict = None)
"(discovered, awaiting_approval, executing, completed, failed, declined, unmapped).",
category="clickup",
)
def clickup_list_tasks(status: str = "", ctx: dict = None) -> str:
def clickup_list_tasks(status: str = "", ctx: dict | None = None) -> str:
"""List tracked ClickUp tasks, optionally filtered by state."""
db = ctx["db"]
states = _get_clickup_states(db)
@ -130,7 +130,7 @@ def clickup_list_tasks(status: str = "", ctx: dict = None) -> str:
"Check the detailed internal processing state of a ClickUp task by its ID.",
category="clickup",
)
def clickup_task_status(task_id: str, ctx: dict = None) -> str:
def clickup_task_status(task_id: str, ctx: dict | None = None) -> str:
"""Get detailed state for a specific tracked task."""
db = ctx["db"]
raw = db.kv_get(f"clickup:task:{task_id}:state")
@ -168,7 +168,7 @@ def clickup_task_status(task_id: str, ctx: dict = None) -> str:
"Approve a ClickUp task that is waiting for permission to execute.",
category="clickup",
)
def clickup_approve_task(task_id: str, ctx: dict = None) -> str:
def clickup_approve_task(task_id: str, ctx: dict | None = None) -> str:
"""Approve a task in awaiting_approval state."""
db = ctx["db"]
key = f"clickup:task:{task_id}:state"
@ -182,11 +182,13 @@ def clickup_approve_task(task_id: str, ctx: dict = None) -> str:
return f"Corrupted state data for task '{task_id}'."
if state.get("state") != "awaiting_approval":
return f"Task '{task_id}' is in state '{state.get('state')}', not 'awaiting_approval'. Cannot approve."
current = state.get("state")
return f"Task '{task_id}' is in state '{current}', not 'awaiting_approval'. Cannot approve."
state["state"] = "approved"
db.kv_set(key, json.dumps(state))
return f"Task '{state.get('clickup_task_name', task_id)}' approved for execution. It will run on the next scheduler cycle."
name = state.get("clickup_task_name", task_id)
return f"Task '{name}' approved for execution. It will run on the next scheduler cycle."
@tool(
@ -194,7 +196,7 @@ def clickup_approve_task(task_id: str, ctx: dict = None) -> str:
"Decline a ClickUp task that is waiting for permission to execute.",
category="clickup",
)
def clickup_decline_task(task_id: str, ctx: dict = None) -> str:
def clickup_decline_task(task_id: str, ctx: dict | None = None) -> str:
"""Decline a task in awaiting_approval state."""
db = ctx["db"]
key = f"clickup:task:{task_id}:state"
@ -208,7 +210,8 @@ def clickup_decline_task(task_id: str, ctx: dict = None) -> str:
return f"Corrupted state data for task '{task_id}'."
if state.get("state") != "awaiting_approval":
return f"Task '{task_id}' is in state '{state.get('state')}', not 'awaiting_approval'. Cannot decline."
current = state.get("state")
return f"Task '{task_id}' is in state '{current}', not 'awaiting_approval'. Cannot decline."
state["state"] = "declined"
db.kv_set(key, json.dumps(state))

10
cheddahbot/tools/data_proc.py
View File

@ -3,7 +3,6 @@
from __future__ import annotations
import csv
import io
import json
from pathlib import Path
@ -34,7 +33,8 @@ def read_csv(path: str, max_rows: int = 20) -> str:
lines.append(" | ".join(str(c)[:50] for c in row))
result = "\n".join(lines)
total_line_count = sum(1 for _ in open(p, encoding="utf-8-sig"))
with open(p, encoding="utf-8-sig") as fcount:
total_line_count = sum(1 for _ in fcount)
if total_line_count > max_rows + 1:
result += f"\n\n... ({total_line_count - 1} total rows, showing first {max_rows})"
return result
@ -66,7 +66,11 @@ def query_json(path: str, json_path: str) -> str:
try:
data = json.loads(p.read_text(encoding="utf-8"))
result = _navigate(data, json_path.split("."))
return json.dumps(result, indent=2, ensure_ascii=False) if not isinstance(result, str) else result
return (
json.dumps(result, indent=2, ensure_ascii=False)
if not isinstance(result, str)
else result
)
except Exception as e:
return f"Error: {e}"

61
cheddahbot/tools/delegate.py
View File

@ -1,14 +1,22 @@
"""Delegate tool: bridges chat brain to execution brain.
"""Delegate tools: bridges between brains and between agents.
When the chat model needs to run commands, edit files, or do anything
requiring system-level access, it calls this tool. The task is passed
to the execution brain (Claude Code CLI) which has full tool access.
delegate_task — sends a task to the execution brain (Claude Code CLI).
delegate_to_agent — routes a task to a named agent in the multi-agent registry.
"""
from __future__ import annotations
import logging
import threading
from . import tool
log = logging.getLogger(__name__)
# Guard against infinite agent-to-agent delegation loops.
_MAX_DELEGATION_DEPTH = 3
_delegation_depth = threading.local()
@tool(
"delegate_task",
@ -21,10 +29,53 @@ from . import tool
),
category="system",
)
def delegate_task(task_description: str, ctx: dict = None) -> str:
def delegate_task(task_description: str, ctx: dict | None = None) -> str:
"""Delegate a task to the execution brain."""
if not ctx or "agent" not in ctx:
return "Error: delegate tool requires agent context."
agent = ctx["agent"]
return agent.execute_task(task_description)
@tool(
"delegate_to_agent",
description=(
"Route a task to a specific named agent. Use this to delegate work to "
"a specialist: e.g. 'researcher' for deep research, 'writer' for content "
"creation, 'ops' for system operations. The target agent processes the "
"task using its own tools, skills, and memory scope, then returns the result."
),
category="system",
)
def delegate_to_agent(
agent_name: str, task_description: str, ctx: dict | None = None
) -> str:
"""Delegate a task to another agent by name."""
if not ctx or "agent_registry" not in ctx:
return "Error: delegate_to_agent requires agent_registry in context."
# Depth guard — prevent infinite A→B→A loops
depth = getattr(_delegation_depth, "value", 0)
if depth >= _MAX_DELEGATION_DEPTH:
return (
f"Error: delegation depth limit ({_MAX_DELEGATION_DEPTH}) reached. "
"Cannot delegate further to prevent infinite loops."
)
registry = ctx["agent_registry"]
target = registry.get(agent_name)
if target is None:
available = ", ".join(registry.list_agents())
return f"Error: agent '{agent_name}' not found. Available agents: {available}"
log.info(
"Delegating to agent '%s' (depth %d): %s",
agent_name, depth + 1, task_description[:100],
)
_delegation_depth.value = depth + 1
try:
return target.respond_to_prompt(task_description)
finally:
_delegation_depth.value = depth

1
cheddahbot/tools/file_ops.py
View File

@ -2,7 +2,6 @@
from __future__ import annotations
import os
from pathlib import Path
from . import tool

21
cheddahbot/tools/image.py
View File

@ -9,14 +9,22 @@ from . import tool
@tool("analyze_image", "Describe or analyze an image file", category="media")
def analyze_image(path: str, question: str = "Describe this image in detail.", ctx: dict = None) -> str:
def analyze_image(
path: str, question: str = "Describe this image in detail.", ctx: dict | None = None
) -> str:
p = Path(path).resolve()
if not p.exists():
return f"Image not found: {path}"
suffix = p.suffix.lower()
mime_map = {".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg",
".gif": "image/gif", ".webp": "image/webp", ".bmp": "image/bmp"}
mime_map = {
".png": "image/png",
".jpg": "image/jpeg",
".jpeg": "image/jpeg",
".gif": "image/gif",
".webp": "image/webp",
".bmp": "image/bmp",
}
mime = mime_map.get(suffix, "image/png")
try:
@ -27,10 +35,13 @@ def analyze_image(path: str, question: str = "Describe this image in detail.", c
if ctx and ctx.get("agent"):
agent = ctx["agent"]
messages = [
{"role": "user", "content": [
{
"role": "user",
"content": [
{"type": "text", "text": question},
{"type": "image_url", "image_url": {"url": f"data:{mime};base64,{data}"}},
]},
],
},
]
result_parts = []
for chunk in agent.llm.chat(messages, stream=False):

155
cheddahbot/tools/press_release.py
View File

@ -3,8 +3,8 @@
Autonomous workflow:
1. Generate 7 compliant headlines (chat brain)
2. AI judge picks the 2 best (chat brain)
3. Write 2 full press releases (execution brain × 2)
4. Generate 2 JSON-LD schemas (execution brain × 2, Sonnet + WebSearch)
3. Write 2 full press releases (execution brain x 2)
4. Generate 2 JSON-LD schemas (execution brain x 2, Sonnet + WebSearch)
5. Save 4 files, return cost summary
"""
@ -14,7 +14,7 @@ import json
import logging
import re
import time
from datetime import datetime
from datetime import UTC, datetime
from pathlib import Path
from ..docx_export import text_to_docx
@ -47,12 +47,21 @@ def _set_status(ctx: dict | None, message: str) -> None:
# Helpers
# ---------------------------------------------------------------------------
def _load_skill(filename: str) -> str:
"""Read a markdown skill file from the skills/ directory."""
"""Read a markdown skill file from the skills/ directory, stripping frontmatter."""
path = _SKILLS_DIR / filename
if not path.exists():
raise FileNotFoundError(f"Skill file not found: {path}")
return path.read_text(encoding="utf-8")
text = path.read_text(encoding="utf-8")
# Strip YAML frontmatter (--- ... ---) if present
if text.startswith("---"):
end = text.find("---", 3)
if end != -1:
text = text[end + 3 :].strip()
return text
def _load_file_if_exists(path: Path) -> str:
@ -137,8 +146,10 @@ def _clean_pr_output(raw: str, headline: str) -> str:
# Prompt builders
# ---------------------------------------------------------------------------
def _build_headline_prompt(topic: str, company_name: str, url: str,
lsi_terms: str, headlines_ref: str) -> str:
def _build_headline_prompt(
topic: str, company_name: str, url: str, lsi_terms: str, headlines_ref: str
) -> str:
"""Build the prompt for Step 1: generate 7 headlines."""
prompt = (
f"Generate exactly 7 unique press release headline options for the following.\n\n"
@ -276,10 +287,17 @@ def _fuzzy_find_anchor(text: str, company_name: str, topic: str) -> str | None:
return None
def _build_pr_prompt(headline: str, topic: str, company_name: str,
url: str, lsi_terms: str, required_phrase: str,
skill_text: str, companies_file: str,
anchor_phrase: str = "") -> str:
def _build_pr_prompt(
headline: str,
topic: str,
company_name: str,
url: str,
lsi_terms: str,
required_phrase: str,
skill_text: str,
companies_file: str,
anchor_phrase: str = "",
) -> str:
"""Build the prompt for Step 3: write one full press release."""
prompt = (
f"{skill_text}\n\n"
@ -299,10 +317,10 @@ def _build_pr_prompt(headline: str, topic: str, company_name: str,
if anchor_phrase:
prompt += (
f'\nANCHOR TEXT REQUIREMENT: You MUST include the exact phrase '
f"\nANCHOR TEXT REQUIREMENT: You MUST include the exact phrase "
f'"{anchor_phrase}" somewhere naturally in the body of the press '
f'release. This phrase will be used as anchor text for an SEO link. '
f'Work it into a sentence where it reads naturally — for example: '
f"release. This phrase will be used as anchor text for an SEO link. "
f"Work it into a sentence where it reads naturally — for example: "
f'"As a {anchor_phrase.split(company_name, 1)[-1].strip()} provider, '
f'{company_name}..." or "{anchor_phrase} continues to...".\n'
)
@ -328,8 +346,7 @@ def _build_pr_prompt(headline: str, topic: str, company_name: str,
return prompt
def _build_schema_prompt(pr_text: str, company_name: str, url: str,
skill_text: str) -> str:
def _build_schema_prompt(pr_text: str, company_name: str, url: str, skill_text: str) -> str:
"""Build the prompt for Step 4: generate JSON-LD schema for one PR."""
prompt = (
f"{skill_text}\n\n"
@ -342,10 +359,7 @@ def _build_schema_prompt(pr_text: str, company_name: str, url: str,
"- No markdown fences, no commentary, no explanations\n"
"- The very first character of your output must be {\n"
)
prompt += (
f"\nCompany name: {company_name}\n\n"
f"Press release text:\n{pr_text}"
)
prompt += f"\nCompany name: {company_name}\n\nPress release text:\n{pr_text}"
return prompt
@ -353,6 +367,7 @@ def _build_schema_prompt(pr_text: str, company_name: str, url: str,
# Main tool
# ---------------------------------------------------------------------------
@tool(
"write_press_releases",
description=(
@ -371,7 +386,7 @@ def write_press_releases(
lsi_terms: str = "",
required_phrase: str = "",
clickup_task_id: str = "",
ctx: dict = None,
ctx: dict | None = None,
) -> str:
"""Run the full press-release pipeline and return results + cost summary."""
if not ctx or "agent" not in ctx:
@ -408,11 +423,13 @@ def write_press_releases(
{"role": "user", "content": headline_prompt},
]
headlines_raw = _chat_call(agent, messages)
cost_log.append({
cost_log.append(
{
"step": "1. Generate 7 headlines",
"model": agent.llm.current_model,
"elapsed_s": round(time.time() - step_start, 1),
})
}
)
if not headlines_raw.strip():
return "Error: headline generation returned empty result."
@ -432,20 +449,36 @@ def write_press_releases(
{"role": "user", "content": judge_prompt},
]
judge_result = _chat_call(agent, messages)
cost_log.append({
cost_log.append(
{
"step": "2. Judge picks best 2",
"model": agent.llm.current_model,
"elapsed_s": round(time.time() - step_start, 1),
})
}
)
# Parse the two winning headlines
winners = [line.strip().lstrip("0123456789.-) ") for line in judge_result.strip().splitlines() if line.strip()]
winners = [
line.strip().lstrip("0123456789.-) ")
for line in judge_result.strip().splitlines()
if line.strip()
]
if len(winners) < 2:
all_headlines = [line.strip().lstrip("0123456789.-) ") for line in headlines_raw.strip().splitlines() if line.strip()]
winners = all_headlines[:2] if len(all_headlines) >= 2 else [all_headlines[0], all_headlines[0]] if all_headlines else ["Headline A", "Headline B"]
all_headlines = [
line.strip().lstrip("0123456789.-) ")
for line in headlines_raw.strip().splitlines()
if line.strip()
]
winners = (
all_headlines[:2]
if len(all_headlines) >= 2
else [all_headlines[0], all_headlines[0]]
if all_headlines
else ["Headline A", "Headline B"]
)
winners = winners[:2]
# ── Step 3: Write 2 press releases (execution brain × 2) ─────────────
# ── Step 3: Write 2 press releases (execution brain x 2) ─────────────
log.info("[PR Pipeline] Step 3/4: Writing 2 press releases...")
anchor_phrase = _derive_anchor_phrase(company_name, topic)
pr_texts: list[str] = []
@ -457,18 +490,26 @@ def write_press_releases(
_set_status(ctx, f"Step 3/4: Writing press release {i + 1}/2 — {headline[:60]}...")
step_start = time.time()
pr_prompt = _build_pr_prompt(
headline, topic, company_name, url, lsi_terms,
required_phrase, pr_skill, companies_file,
headline,
topic,
company_name,
url,
lsi_terms,
required_phrase,
pr_skill,
companies_file,
anchor_phrase=anchor_phrase,
)
exec_tools = "Bash,Read,Edit,Write,Glob,Grep,WebFetch"
raw_result = agent.execute_task(pr_prompt, tools=exec_tools)
elapsed = round(time.time() - step_start, 1)
cost_log.append({
cost_log.append(
{
"step": f"3{chr(97 + i)}. Write PR '{headline[:40]}...'",
"model": "execution-brain (default)",
"elapsed_s": elapsed,
})
}
)
# Clean output: find the headline, strip preamble and markdown
clean_result = _clean_pr_output(raw_result, headline)
@ -487,13 +528,13 @@ def write_press_releases(
if fuzzy:
log.info("PR %d: exact anchor not found, fuzzy match: '%s'", i + 1, fuzzy)
anchor_warnings.append(
f"PR {chr(65+i)}: Exact anchor phrase \"{anchor_phrase}\" not found. "
f"Closest match: \"{fuzzy}\" — you may want to adjust before submitting."
f'PR {chr(65 + i)}: Exact anchor phrase "{anchor_phrase}" not found. '
f'Closest match: "{fuzzy}" — you may want to adjust before submitting.'
)
else:
log.warning("PR %d: anchor phrase '%s' NOT found", i + 1, anchor_phrase)
anchor_warnings.append(
f"PR {chr(65+i)}: Anchor phrase \"{anchor_phrase}\" NOT found in the text. "
f'PR {chr(65 + i)}: Anchor phrase "{anchor_phrase}" NOT found in the text. '
f"You'll need to manually add it before submitting to PA."
)
@ -525,11 +566,13 @@ def write_press_releases(
model=SONNET_CLI_MODEL,
)
elapsed = round(time.time() - step_start, 1)
cost_log.append({
cost_log.append(
{
"step": f"4{chr(97 + i)}. Schema for PR {i + 1}",
"model": SONNET_CLI_MODEL,
"elapsed_s": elapsed,
})
}
)
# Extract clean JSON and force correct mainEntityOfPage
schema_json = _extract_json(result)
@ -573,7 +616,7 @@ def write_press_releases(
# Anchor text warnings
if anchor_warnings:
output_parts.append("## Anchor Text Warnings\n")
output_parts.append(f"Required anchor phrase: **\"{anchor_phrase}\"**\n")
output_parts.append(f'Required anchor phrase: **"{anchor_phrase}"**\n')
for warning in anchor_warnings:
output_parts.append(f"- {warning}")
output_parts.append("")
@ -608,10 +651,11 @@ def write_press_releases(
# Post a result comment
attach_note = f"\n📎 {uploaded_count} file(s) attached." if uploaded_count else ""
result_text = "\n".join(output_parts)[:3000]
comment = (
f"✅ CheddahBot completed this task (via chat).\n\n"
f"Skill: write_press_releases\n"
f"Result:\n{'\n'.join(output_parts)[:3000]}{attach_note}"
f"Result:\n{result_text}{attach_note}"
)
client.add_comment(clickup_task_id, comment)
@ -622,19 +666,19 @@ def write_press_releases(
db = ctx.get("db")
if db:
import json as _json
kv_key = f"clickup:task:{clickup_task_id}:state"
existing = db.kv_get(kv_key)
if existing:
from datetime import timezone
state = _json.loads(existing)
state["state"] = "completed"
state["completed_at"] = datetime.now(timezone.utc).isoformat()
state["completed_at"] = datetime.now(UTC).isoformat()
state["deliverable_paths"] = docx_files
db.kv_set(kv_key, _json.dumps(state))
client.close()
output_parts.append(f"\n## ClickUp Sync\n")
output_parts.append("\n## ClickUp Sync\n")
output_parts.append(f"- Task `{clickup_task_id}` updated")
output_parts.append(f"- {uploaded_count} file(s) uploaded")
output_parts.append(f"- Status set to '{config.clickup.review_status}'")
@ -642,7 +686,7 @@ def write_press_releases(
log.info("ClickUp sync complete for task %s", clickup_task_id)
except Exception as e:
log.error("ClickUp sync failed for task %s: %s", clickup_task_id, e)
output_parts.append(f"\n## ClickUp Sync\n")
output_parts.append("\n## ClickUp Sync\n")
output_parts.append(f"- **Sync failed:** {e}")
output_parts.append("- Press release results are still valid above")
@ -683,7 +727,7 @@ def _parse_company_data(companies_text: str) -> dict[str, dict]:
current_data = {"name": current_company}
elif current_company:
if line.startswith("- **PA Org ID:**"):
try:
try: # noqa: SIM105
current_data["org_id"] = int(line.split(":**")[1].strip())
except (ValueError, IndexError):
pass
@ -811,13 +855,14 @@ def _extract_json(text: str) -> str | None:
except json.JSONDecodeError:
pass
return None # noqa: RET501
return None
# ---------------------------------------------------------------------------
# Submit tool
# ---------------------------------------------------------------------------
def _resolve_branded_url(branded_url: str, company_data: dict | None) -> str:
"""Resolve the branded link URL.
@ -867,12 +912,12 @@ def _build_links(
if fuzzy:
links.append({"url": target_url, "anchor": fuzzy})
warnings.append(
f"Brand+keyword link: exact phrase \"{anchor_phrase}\" not found. "
f"Used fuzzy match: \"{fuzzy}\""
f'Brand+keyword link: exact phrase "{anchor_phrase}" not found. '
f'Used fuzzy match: "{fuzzy}"'
)
else:
warnings.append(
f"Brand+keyword link: anchor phrase \"{anchor_phrase}\" NOT found in PR text. "
f'Brand+keyword link: anchor phrase "{anchor_phrase}" NOT found in PR text. '
f"Link to {target_url} could not be injected — add it manually in PA."
)
@ -883,7 +928,7 @@ def _build_links(
links.append({"url": branded_url_resolved, "anchor": company_name})
else:
warnings.append(
f"Branded link: company name \"{company_name}\" not found in PR text. "
f'Branded link: company name "{company_name}" not found in PR text. '
f"Link to {branded_url_resolved} could not be injected."
)
@ -911,7 +956,7 @@ def submit_press_release(
pr_text: str = "",
file_path: str = "",
description: str = "",
ctx: dict = None,
ctx: dict | None = None,
) -> str:
"""Submit a finished press release to Press Advantage as a draft."""
# --- Get config ---
@ -991,7 +1036,11 @@ def submit_press_release(
# --- Build links ---
branded_url_resolved = _resolve_branded_url(branded_url, company_data)
link_list, link_warnings = _build_links(
pr_text, company_name, topic, target_url, branded_url_resolved,
pr_text,
company_name,
topic,
target_url,
branded_url_resolved,
)
# --- Convert to HTML ---
@ -1039,7 +1088,7 @@ def submit_press_release(
if link_list:
output_parts.append("\n**Links:**")
for link in link_list:
output_parts.append(f" - \"{link['anchor']}\" → {link['url']}")
output_parts.append(f' - "{link["anchor"]}" → {link["url"]}')
if link_warnings:
output_parts.append("\n**Link warnings:**")

10
cheddahbot/tools/shell.py
View File

@ -3,7 +3,6 @@
from __future__ import annotations
import subprocess
import sys
from . import tool
@ -19,7 +18,14 @@ BLOCKED_PATTERNS = [
@tool("run_command", "Execute a shell command and return output", category="shell")
def run_command(command: str, timeout: int = 30) -> str:
def run_command(command: str, timeout: int = 30, ctx: dict | None = None) -> str:
# Check require_approval setting
if ctx and ctx.get("config") and ctx["config"].shell.require_approval:
return (
"Shell commands require approval. Use the `delegate_task` tool instead — "
"it routes through the execution brain which has its own safety controls."
)
# Safety check
cmd_lower = command.lower().strip()
for pattern in BLOCKED_PATTERNS:

2
cheddahbot/tools/web.py
View File

@ -51,7 +51,7 @@ def fetch_url(url: str) -> str:
tag.decompose()
text = soup.get_text(separator="\n", strip=True)
# Collapse whitespace
lines = [l.strip() for l in text.split("\n") if l.strip()]
lines = [line.strip() for line in text.split("\n") if line.strip()]
text = "\n".join(lines)
if len(text) > 15000:
text = text[:15000] + "\n... (truncated)"

81
cheddahbot/ui.py
View File

@ -16,8 +16,9 @@ if TYPE_CHECKING:
log = logging.getLogger(__name__)
_HEAD = '<meta name="viewport" content="width=device-width, initial-scale=1">'
_CSS = """
.contain { max-width: 900px; margin: auto; }
footer { display: none !important; }
.notification-banner {
background: #1a1a2e;
@ -27,11 +28,54 @@ footer { display: none !important; }
margin-bottom: 8px;
font-size: 0.9em;
}
/* Mobile optimizations */
@media (max-width: 768px) {
.gradio-container { padding: 4px !important; }
/* 16px base font on chat messages to prevent iOS zoom on focus */
.chatbot .message-row .message { font-size: 16px !important; }
/* Chat container: scrollable, no zoom-stuck overflow */
.chatbot {
overflow-y: auto !important;
-webkit-overflow-scrolling: touch;
height: calc(100dvh - 220px) !important;
max-height: none !important;
}
/* Tighten up header/status bar spacing */
.gradio-container > .main > .wrap { gap: 8px !important; }
/* Keep input area pinned at the bottom, never overlapping chat */
.gradio-container > .main {
display: flex;
flex-direction: column;
height: 100dvh;
}
.gradio-container > .main > .wrap:last-child {
position: sticky;
bottom: 0;
background: var(--background-fill-primary);
padding-bottom: env(safe-area-inset-bottom, 8px);
z-index: 10;
}
/* Input box: prevent tiny text that triggers zoom */
.multimodal-textbox textarea,
.multimodal-textbox input {
font-size: 16px !important;
}
/* Reduce model dropdown row padding */
.contain .gr-row { gap: 4px !important; }
}
"""
def create_ui(agent: Agent, config: Config, llm: LLMAdapter,
notification_bus: NotificationBus | None = None) -> gr.Blocks:
def create_ui(
agent: Agent, config: Config, llm: LLMAdapter, notification_bus: NotificationBus | None = None
) -> gr.Blocks:
"""Build and return the Gradio app."""
available_models = llm.list_chat_models()
@ -41,7 +85,7 @@ def create_ui(agent: Agent, config: Config, llm: LLMAdapter,
exec_status = "available" if llm.is_execution_brain_available() else "unavailable"
clickup_status = "enabled" if config.clickup.enabled else "disabled"
with gr.Blocks(title="CheddahBot") as app:
with gr.Blocks(title="CheddahBot", fill_width=True, css=_CSS, head=_HEAD) as app:
gr.Markdown("# CheddahBot", elem_classes=["contain"])
gr.Markdown(
f"*Chat Brain:* `{current_model}` &nbsp;|&nbsp; "
@ -90,7 +134,6 @@ def create_ui(agent: Agent, config: Config, llm: LLMAdapter,
sources=["upload", "microphone"],
)
# -- Event handlers --
def on_model_change(model_id):
@ -125,12 +168,23 @@ def create_ui(agent: Agent, config: Config, llm: LLMAdapter,
processed_files = []
for f in files:
fpath = f if isinstance(f, str) else f.get("path", f.get("name", ""))
if fpath and Path(fpath).suffix.lower() in (".wav", ".mp3", ".ogg", ".webm", ".m4a"):
if fpath and Path(fpath).suffix.lower() in (
".wav",
".mp3",
".ogg",
".webm",
".m4a",
):
try:
from .media import transcribe_audio
transcript = transcribe_audio(fpath)
if transcript:
text = f"{text}\n[Voice message]: {transcript}" if text else f"[Voice message]: {transcript}"
text = (
f"{text}\n[Voice message]: {transcript}"
if text
else f"[Voice message]: {transcript}"
)
continue
except Exception as e:
log.warning("Audio transcription failed: %s", e)
@ -142,13 +196,13 @@ def create_ui(agent: Agent, config: Config, llm: LLMAdapter,
file_names = [Path(f).name for f in processed_files]
user_display += f"\n[Attached: {', '.join(file_names)}]"
chat_history = chat_history + [{"role": "user", "content": user_display}]
chat_history = [*chat_history, {"role": "user", "content": user_display}]
yield chat_history, gr.update(value=None)
# Stream assistant response
try:
response_text = ""
chat_history = chat_history + [{"role": "assistant", "content": ""}]
chat_history = [*chat_history, {"role": "assistant", "content": ""}]
for chunk in agent.respond(text, files=processed_files):
response_text += chunk
@ -157,11 +211,14 @@ def create_ui(agent: Agent, config: Config, llm: LLMAdapter,
# If no response came through, show a fallback
if not response_text:
chat_history[-1] = {"role": "assistant", "content": "(No response received from model)"}
chat_history[-1] = {
"role": "assistant",
"content": "(No response received from model)",
}
yield chat_history, gr.update(value=None)
except Exception as e:
log.error("Error in agent.respond: %s", e, exc_info=True)
chat_history = chat_history + [{"role": "assistant", "content": f"Error: {e}"}]
chat_history = [*chat_history, {"role": "assistant", "content": f"Error: {e}"}]
yield chat_history, gr.update(value=None)
def poll_pipeline_status():
@ -209,4 +266,4 @@ def create_ui(agent: Agent, config: Config, llm: LLMAdapter,
timer = gr.Timer(10)
timer.tick(poll_notifications, None, [notification_display])
return app, _CSS
return app

29
config.yaml
View File

@ -56,3 +56,32 @@ clickup:
company_name: "Client"
target_url: "IMSURL"
branded_url: "SocialURL"
# Multi-agent configuration
# Each agent gets its own personality, tool whitelist, and memory scope.
# The first agent is the default. Omit this section for single-agent mode.
agents:
- name: default
display_name: CheddahBot
# tools: null = all tools, [] = no tools
# skills: null = auto (all skills matching agent name)
# memory_scope: "" = shared memory
- name: writer
display_name: Writing Agent
personality_file: "" # future: identity/WRITER.md
skills: [press-release-writer, press-release-schema]
tools: [write_press_releases, submit_press_release, delegate_task, remember, search_memory]
memory_scope: "" # shares memory with default
- name: researcher
display_name: Research Agent
personality_file: "" # future: identity/RESEARCHER.md
tools: [web_search, web_fetch, delegate_task, remember, search_memory]
memory_scope: ""
- name: ops
display_name: Ops Agent
personality_file: "" # future: identity/OPS.md
tools: [run_command, delegate_task, list_files, read_file, remember, search_memory]
memory_scope: ""

2
skills/companies.md
View File

@ -114,7 +114,7 @@
- **Website:**
- **GBP:**
## FZE Industrial
## FZE Manufacturing
- **Executive:** Doug Pribyl, CEO
- **PA Org ID:** 22377
- **Website:**

2
skills/press-release-schema.md
View File

@ -1,6 +1,8 @@
---
name: press-release-schema
description: Generate valid NewsArticle JSON-LD schema markup for press releases with proper entity linking to Wikipedia. UPDATED VERSION with fixed entity identification. Use when the user asks to create, generate, or add schema markup, structured data, JSON-LD, or SEO schema for a press release, news article, or announcement.
tools: [write_press_releases]
agents: [writer, default]
---
# Press Release Schema Generator v2

2
skills/press_release_prompt.md
View File

@ -1,6 +1,8 @@
---
name: press-release-writer
description: Professional press release writing that follows Press Advantage guidelines and journalistic standards. Use when the user asks to write a press release, create a news announcement, draft a PR, or mentions Press Advantage distribution. Automatically generates LSI terms and industry entities, follows strict formatting rules (no lists/bullets/questions/headings in body, third-person only), and produces 600-750 word releases in objective journalistic style.
tools: [write_press_releases, submit_press_release]
agents: [writer, default]
---
# Press Release Writer

3
tests/conftest.py
View File

@ -2,9 +2,6 @@
from __future__ import annotations
import tempfile
from pathlib import Path
import pytest
from cheddahbot.db import Database

13
tests/test_clickup.py
View File

@ -7,7 +7,6 @@ import respx
from cheddahbot.clickup import BASE_URL, ClickUpClient, ClickUpTask
# ── ClickUpTask.from_api ──
@ -183,9 +182,7 @@ class TestClickUpClient:
@respx.mock
def test_update_task_status(self):
respx.put(f"{BASE_URL}/task/t1").mock(
return_value=httpx.Response(200, json={})
)
respx.put(f"{BASE_URL}/task/t1").mock(return_value=httpx.Response(200, json={}))
client = ClickUpClient(api_token="pk_test_123")
result = client.update_task_status("t1", "in progress")
@ -210,9 +207,7 @@ class TestClickUpClient:
@respx.mock
def test_add_comment(self):
respx.post(f"{BASE_URL}/task/t1/comment").mock(
return_value=httpx.Response(200, json={})
)
respx.post(f"{BASE_URL}/task/t1/comment").mock(return_value=httpx.Response(200, json={}))
client = ClickUpClient(api_token="pk_test_123")
result = client.add_comment("t1", "CheddahBot completed this task.")
@ -260,9 +255,7 @@ class TestClickUpClient:
docx_file = tmp_path / "report.docx"
docx_file.write_bytes(b"fake docx content")
respx.post(f"{BASE_URL}/task/t1/attachment").mock(
return_value=httpx.Response(200, json={})
)
respx.post(f"{BASE_URL}/task/t1/attachment").mock(return_value=httpx.Response(200, json={}))
client = ClickUpClient(api_token="pk_test_123")
result = client.upload_attachment("t1", docx_file)

2
tests/test_clickup_tools.py
View File

@ -4,8 +4,6 @@ from __future__ import annotations
import json
import pytest
from cheddahbot.tools.clickup_tool import (
clickup_approve_task,
clickup_decline_task,

4
tests/test_db.py
View File

@ -64,8 +64,8 @@ class TestNotifications:
def test_after_id_filters_correctly(self, tmp_db):
id1 = tmp_db.add_notification("First", "clickup")
id2 = tmp_db.add_notification("Second", "clickup")
id3 = tmp_db.add_notification("Third", "clickup")
_id2 = tmp_db.add_notification("Second", "clickup")
_id3 = tmp_db.add_notification("Third", "clickup")
# Should only get notifications after id1
notifs = tmp_db.get_notifications_after(id1)

117
tests/test_press_advantage.py
View File

@ -2,7 +2,6 @@
from __future__ import annotations
from pathlib import Path
from unittest.mock import MagicMock
import httpx
@ -24,7 +23,6 @@ from cheddahbot.tools.press_release import (
submit_press_release,
)
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@ -81,19 +79,21 @@ def submit_ctx(pa_config):
# PressAdvantageClient tests
# ---------------------------------------------------------------------------
class TestPressAdvantageClient:
class TestPressAdvantageClient:
@respx.mock
def test_get_organizations(self):
respx.get(
"https://app.pressadvantage.com/api/customers/organizations.json",
).mock(return_value=httpx.Response(
).mock(
return_value=httpx.Response(
200,
json=[
{"id": 19634, "name": "Advanced Industrial"},
{"id": 19800, "name": "Metal Craft"},
],
))
)
)
client = PressAdvantageClient("test-key")
try:
@ -108,10 +108,12 @@ class TestPressAdvantageClient:
def test_create_release_success(self):
respx.post(
"https://app.pressadvantage.com/api/customers/releases/with_content.json",
).mock(return_value=httpx.Response(
).mock(
return_value=httpx.Response(
200,
json={"id": 99999, "state": "draft", "title": "Test Headline"},
))
)
)
client = PressAdvantageClient("test-key")
try:
@ -154,10 +156,12 @@ class TestPressAdvantageClient:
def test_get_release(self):
respx.get(
"https://app.pressadvantage.com/api/customers/releases/81505.json",
).mock(return_value=httpx.Response(
).mock(
return_value=httpx.Response(
200,
json={"id": 81505, "state": "draft", "title": "Test"},
))
)
)
client = PressAdvantageClient("test-key")
try:
@ -171,10 +175,12 @@ class TestPressAdvantageClient:
def test_get_built_urls(self):
respx.get(
"https://app.pressadvantage.com/api/customers/releases/81505/built_urls.json",
).mock(return_value=httpx.Response(
).mock(
return_value=httpx.Response(
200,
json=[{"url": "https://example.com/press-release"}],
))
)
)
client = PressAdvantageClient("test-key")
try:
@ -204,6 +210,7 @@ class TestPressAdvantageClient:
# Company data parsing tests
# ---------------------------------------------------------------------------
class TestParseCompanyOrgIds:
def test_parses_all_companies(self):
mapping = _parse_company_org_ids(SAMPLE_COMPANIES_MD)
@ -280,12 +287,19 @@ class TestFuzzyMatchCompanyData:
# Anchor phrase helpers
# ---------------------------------------------------------------------------
class TestDeriveAnchorPhrase:
def test_basic(self):
assert _derive_anchor_phrase("Advanced Industrial", "PEEK machining") == "Advanced Industrial PEEK machining"
assert (
_derive_anchor_phrase("Advanced Industrial", "PEEK machining")
== "Advanced Industrial PEEK machining"
)
def test_strips_whitespace(self):
assert _derive_anchor_phrase("Metal Craft", " custom fabrication ") == "Metal Craft custom fabrication"
assert (
_derive_anchor_phrase("Metal Craft", " custom fabrication ")
== "Metal Craft custom fabrication"
)
class TestFindAnchorInText:
@ -325,10 +339,14 @@ class TestFuzzyFindAnchor:
# Branded URL resolution
# ---------------------------------------------------------------------------
class TestResolveBrandedUrl:
def test_literal_url(self):
data = {"website": "https://example.com", "gbp": "https://maps.google.com/123"}
assert _resolve_branded_url("https://linkedin.com/company/acme", data) == "https://linkedin.com/company/acme"
assert (
_resolve_branded_url("https://linkedin.com/company/acme", data)
== "https://linkedin.com/company/acme"
)
def test_gbp_shortcut(self):
data = {"website": "https://example.com", "gbp": "https://maps.google.com/maps?cid=123"}
@ -358,12 +376,16 @@ class TestResolveBrandedUrl:
# Link building
# ---------------------------------------------------------------------------
class TestBuildLinks:
def test_both_links_found(self):
text = "Advanced Industrial PEEK machining is excellent. Advanced Industrial leads the way."
links, warnings = _build_links(
text, "Advanced Industrial", "PEEK machining",
"https://example.com/peek", "https://linkedin.com/company/ai",
text,
"Advanced Industrial",
"PEEK machining",
"https://example.com/peek",
"https://linkedin.com/company/ai",
)
assert len(links) == 2
assert links[0]["url"] == "https://example.com/peek"
@ -380,9 +402,12 @@ class TestBuildLinks:
def test_brand_keyword_not_found_warns(self):
text = "This text has no relevant anchor phrases at all. " * 30
links, warnings = _build_links(
text, "Advanced Industrial", "PEEK machining",
"https://example.com/peek", "",
_links, warnings = _build_links(
text,
"Advanced Industrial",
"PEEK machining",
"https://example.com/peek",
"",
)
assert len(warnings) == 1
assert "NOT found" in warnings[0]
@ -390,8 +415,11 @@ class TestBuildLinks:
def test_fuzzy_match_used(self):
text = "Advanced Industrial provides excellent PEEK solutions to many clients worldwide."
links, warnings = _build_links(
text, "Advanced Industrial", "PEEK machining",
"https://example.com/peek", "",
text,
"Advanced Industrial",
"PEEK machining",
"https://example.com/peek",
"",
)
# Fuzzy should find "Advanced Industrial provides excellent PEEK" or similar
assert len(links) == 1
@ -404,6 +432,7 @@ class TestBuildLinks:
# Text to HTML
# ---------------------------------------------------------------------------
class TestTextToHtml:
def test_basic_paragraphs(self):
text = "First paragraph.\n\nSecond paragraph."
@ -451,12 +480,15 @@ class TestTextToHtml:
# submit_press_release tool tests
# ---------------------------------------------------------------------------
class TestSubmitPressRelease:
def test_missing_api_key(self):
config = MagicMock()
config.press_advantage.api_key = ""
result = submit_press_release(
headline="Test", company_name="Acme", pr_text=LONG_PR_TEXT,
headline="Test",
company_name="Acme",
pr_text=LONG_PR_TEXT,
ctx={"config": config},
)
assert "PRESS_ADVANTAGE_API" in result
@ -464,13 +496,16 @@ class TestSubmitPressRelease:
def test_missing_context(self):
result = submit_press_release(
headline="Test", company_name="Acme", pr_text=LONG_PR_TEXT,
headline="Test",
company_name="Acme",
pr_text=LONG_PR_TEXT,
)
assert "Error" in result
def test_no_pr_text_or_file(self, submit_ctx):
result = submit_press_release(
headline="Test", company_name="Advanced Industrial",
headline="Test",
company_name="Advanced Industrial",
ctx=submit_ctx,
)
assert "Error" in result
@ -479,16 +514,20 @@ class TestSubmitPressRelease:
def test_word_count_too_low(self, submit_ctx):
short_text = " ".join(["word"] * 100)
result = submit_press_release(
headline="Test", company_name="Advanced Industrial",
pr_text=short_text, ctx=submit_ctx,
headline="Test",
company_name="Advanced Industrial",
pr_text=short_text,
ctx=submit_ctx,
)
assert "Error" in result
assert "550 words" in result
def test_file_not_found(self, submit_ctx):
result = submit_press_release(
headline="Test", company_name="Advanced Industrial",
file_path="/nonexistent/file.txt", ctx=submit_ctx,
headline="Test",
company_name="Advanced Industrial",
file_path="/nonexistent/file.txt",
ctx=submit_ctx,
)
assert "Error" in result
assert "file not found" in result
@ -502,10 +541,12 @@ class TestSubmitPressRelease:
respx.post(
"https://app.pressadvantage.com/api/customers/releases/with_content.json",
).mock(return_value=httpx.Response(
).mock(
return_value=httpx.Response(
200,
json={"id": 88888, "state": "draft"},
))
)
)
result = submit_press_release(
headline="Advanced Industrial Expands PEEK Machining",
@ -526,7 +567,7 @@ class TestSubmitPressRelease:
lambda p: SAMPLE_COMPANIES_MD,
)
route = respx.post(
respx.post(
"https://app.pressadvantage.com/api/customers/releases/with_content.json",
).mock(return_value=httpx.Response(200, json={"id": 1, "state": "draft"}))
@ -549,7 +590,7 @@ class TestSubmitPressRelease:
lambda p: SAMPLE_COMPANIES_MD,
)
route = respx.post(
respx.post(
"https://app.pressadvantage.com/api/customers/releases/with_content.json",
).mock(return_value=httpx.Response(200, json={"id": 1, "state": "draft"}))
@ -599,8 +640,10 @@ class TestSubmitPressRelease:
).mock(return_value=httpx.Response(200, json=[]))
result = submit_press_release(
headline="Test", company_name="Totally Unknown Corp",
pr_text=LONG_PR_TEXT, ctx=submit_ctx,
headline="Test",
company_name="Totally Unknown Corp",
pr_text=LONG_PR_TEXT,
ctx=submit_ctx,
)
assert "Error" in result
@ -615,10 +658,12 @@ class TestSubmitPressRelease:
respx.get(
"https://app.pressadvantage.com/api/customers/organizations.json",
).mock(return_value=httpx.Response(
).mock(
return_value=httpx.Response(
200,
json=[{"id": 12345, "name": "New Client Co"}],
))
)
)
respx.post(
"https://app.pressadvantage.com/api/customers/releases/with_content.json",

6
tests/test_scheduler_helpers.py
View File

@ -26,11 +26,7 @@ class TestExtractDocxPaths:
assert paths == []
def test_only_matches_docx_extension(self):
result = (
"**Docx:** `report.docx`\n"
"**PDF:** `report.pdf`\n"
"**Docx:** `summary.txt`\n"
)
result = "**Docx:** `report.docx`\n**PDF:** `report.pdf`\n**Docx:** `summary.txt`\n"
paths = _extract_docx_paths(result)
assert paths == ["report.docx"]