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

Steps 7-8: Eliminate KV store from task pipelines and update tests 

Remove all KV store reads/writes from task pipeline code. ClickUp is now
the single source of truth for task state. File location (processed/
subfolder) tracks file processing state. Loop timestamps use in-memory
dict on Scheduler.

Source changes:
- scheduler.py: Remove KV dedup, fallback sync path, docx extraction;
  tools own their ClickUp sync; in-memory timestamps
- press_release.py: Remove KV state writes, log-only _set_status
- linkbuilding.py: Remove KV state writes, processed/ subfolder check
- content_creation.py: Phase detection via ClickUp API status, remove
  KV phase/state tracking, _update_kv_state removed
- clickup_tool.py: Rewrite to query ClickUp API directly
- ui.py: Pipeline status polling is now a no-op

Test changes:
- test_scheduler.py: Remove KV dedup tests, remove fallback path test,
  verify ClickUp API calls instead of KV state
- test_content_creation.py: Mock _get_clickup_client for phase detection,
  verify ClickUp sync calls instead of KV assertions
- test_linkbuilding.py: Remove KV status test, verify ClickUp API calls
- test_clickup_tools.py: Rewrite for API-backed tools
- test_scheduler_helpers.py: Test in-memory timestamps

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
fix/customer-field-migration
PeninsulaInd 2026-02-27 16:26:44 -06:00
parent 9a3ba54974
commit 917445ade4
11 changed files with 458 additions and 920 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

332
cheddahbot/scheduler.py
View File

@ -3,7 +3,6 @@
from __future__ import annotations
import contextlib
import json
import logging
import re
import shutil
@ -24,15 +23,6 @@ log = logging.getLogger(__name__)
HEARTBEAT_OK = "HEARTBEAT_OK"
# Matches **Docx:** `path/to/file.docx` patterns in tool output
_DOCX_PATH_RE = re.compile(r"\*\*Docx:\*\*\s*`([^`]+\.docx)`")
def _extract_docx_paths(result: str) -> list[str]:
"""Extract .docx file paths from a tool result string."""
return _DOCX_PATH_RE.findall(result)
class Scheduler:
# Tasks due within this window are eligible for execution
DUE_DATE_WINDOW_WEEKS = 3
@ -60,6 +50,14 @@ class Scheduler:
self._force_autocora = threading.Event()
self._clickup_client = None
self._field_filter_cache: dict | None = None
self._loop_timestamps: dict[str, str | None] = {
"heartbeat": None,
"poll": None,
"clickup": None,
"folder_watch": None,
"autocora": None,
"content_watch": None,
}
def start(self):
"""Start the scheduler, heartbeat, and ClickUp threads."""
@ -169,15 +167,8 @@ class Scheduler:
self._force_autocora.set()
def get_loop_timestamps(self) -> dict[str, str | None]:
"""Return last_run timestamps for all loops."""
return {
"heartbeat": self.db.kv_get("system:loop:heartbeat:last_run"),
"poll": self.db.kv_get("system:loop:poll:last_run"),
"clickup": self.db.kv_get("system:loop:clickup:last_run"),
"folder_watch": self.db.kv_get("system:loop:folder_watch:last_run"),
"autocora": self.db.kv_get("system:loop:autocora:last_run"),
"content_watch": self.db.kv_get("system:loop:content_watch:last_run"),
}
"""Return last_run timestamps for all loops (in-memory)."""
return dict(self._loop_timestamps)
# ── Scheduled Tasks ──
@ -185,9 +176,7 @@ class Scheduler:
while not self._stop_event.is_set():
try:
self._run_due_tasks()
self.db.kv_set(
"system:loop:poll:last_run", datetime.now(UTC).isoformat()
)
self._loop_timestamps["poll"] = datetime.now(UTC).isoformat()
except Exception as e:
log.error("Scheduler poll error: %s", e)
self._interruptible_wait(
@ -227,9 +216,7 @@ class Scheduler:
while not self._stop_event.is_set():
try:
self._run_heartbeat()
self.db.kv_set(
"system:loop:heartbeat:last_run", datetime.now(UTC).isoformat()
)
self._loop_timestamps["heartbeat"] = datetime.now(UTC).isoformat()
except Exception as e:
log.error("Heartbeat error: %s", e)
self._interruptible_wait(interval, self._force_heartbeat)
@ -281,9 +268,7 @@ class Scheduler:
try:
self._poll_clickup()
self._recover_stale_tasks()
self.db.kv_set(
"system:loop:clickup:last_run", datetime.now(UTC).isoformat()
)
self._loop_timestamps["clickup"] = datetime.now(UTC).isoformat()
except Exception as e:
log.error("ClickUp poll error: %s", e)
self._interruptible_wait(interval)
@ -357,15 +342,9 @@ class Scheduler:
)
for task in tasks:
# Skip tasks already processed in kv_store
raw = self.db.kv_get(f"clickup:task:{task.id}:state")
if raw:
try:
existing = json.loads(raw)
if existing.get("state") in ("executing", "completed", "failed"):
continue
except json.JSONDecodeError:
pass
# ClickUp status filtering is the dedup: tasks in poll_statuses
# are eligible; once moved to "automation underway", they won't
# appear in the next poll.
# Client-side verify: Work Category must be in skill_map
if task.task_type not in skill_map:
@ -394,7 +373,11 @@ class Scheduler:
self._execute_task(task)
def _execute_task(self, task):
"""Execute a single ClickUp task immediately."""
"""Execute a single ClickUp task immediately.
Tools own their own ClickUp sync (status, comments, attachments).
The scheduler just calls the tool and handles errors.
"""
skill_map = self.config.clickup.skill_map
mapping = skill_map.get(task.task_type, {})
tool_name = mapping.get("tool", "")
@ -403,35 +386,17 @@ class Scheduler:
return
task_id = task.id
kv_key = f"clickup:task:{task_id}:state"
now = datetime.now(UTC).isoformat()
client = self._get_clickup_client()
# Build state object — starts at "executing"
state = {
"state": "executing",
"clickup_task_id": task_id,
"clickup_task_name": task.name,
"task_type": task.task_type,
"skill_name": tool_name,
"discovered_at": now,
"started_at": now,
"completed_at": None,
"error": None,
"deliverable_paths": [],
"custom_fields": task.custom_fields,
}
# Move to "automation underway" on ClickUp immediately
client.update_task_status(task_id, self.config.clickup.automation_status)
self.db.kv_set(kv_key, json.dumps(state))
log.info("Executing ClickUp task: %s → %s", task.name, tool_name)
self._notify(f"Executing ClickUp task: **{task.name}** → Skill: `{tool_name}`")
try:
# Build tool arguments from field mapping
args = self._build_tool_args(state)
args = self._build_tool_args_from_task(task, mapping)
args["clickup_task_id"] = task_id
# Execute the skill via the tool registry
@ -440,21 +405,15 @@ class Scheduler:
else:
result = self.agent.execute_task(
f"Execute the '{tool_name}' tool for ClickUp task '{task.name}'. "
f"Task description: {state.get('custom_fields', {})}"
f"Task description: {task.custom_fields}"
)
# Check if the tool skipped or reported an error without doing work
if result.startswith("Skipped:") or result.startswith("Error:"):
state["state"] = "failed"
state["error"] = result[:500]
state["completed_at"] = datetime.now(UTC).isoformat()
self.db.kv_set(kv_key, json.dumps(state))
client.add_comment(
task_id,
f"⚠️ CheddahBot could not execute this task.\n\n{result[:2000]}",
)
# Move to "error" so Bryan can see what happened
client.update_task_status(task_id, self.config.clickup.error_status)
self._notify(
@ -464,54 +423,17 @@ class Scheduler:
log.info("ClickUp task skipped: %s — %s", task.name, result[:200])
return
# Check if the tool already handled ClickUp sync internally
tool_handled_sync = "## ClickUp Sync" in result
if tool_handled_sync:
state["state"] = "completed"
state["completed_at"] = datetime.now(UTC).isoformat()
self.db.kv_set(kv_key, json.dumps(state))
else:
# Scheduler handles sync (fallback path)
docx_paths = _extract_docx_paths(result)
state["deliverable_paths"] = docx_paths
uploaded_count = 0
for path in docx_paths:
if client.upload_attachment(task_id, path):
uploaded_count += 1
else:
log.warning("Failed to upload %s for task %s", path, task_id)
state["state"] = "completed"
state["completed_at"] = datetime.now(UTC).isoformat()
self.db.kv_set(kv_key, json.dumps(state))
client.update_task_status(task_id, self.config.clickup.review_status)
attach_note = f"\n📎 {uploaded_count} file(s) attached." if uploaded_count else ""
comment = (
f"✅ CheddahBot completed this task.\n\n"
f"Skill: {tool_name}\n"
f"Result:\n{result[:3000]}{attach_note}"
)
client.add_comment(task_id, comment)
# Tool handled its own ClickUp sync — just log success
self._notify(
f"ClickUp task completed: **{task.name}**\n"
f"Skill: `{tool_name}` | Status set to '{self.config.clickup.review_status}'"
f"Skill: `{tool_name}`"
)
log.info("ClickUp task completed: %s", task.name)
except Exception as e:
# Failure — move back to "to do" on ClickUp
state["state"] = "failed"
state["error"] = str(e)
state["completed_at"] = datetime.now(UTC).isoformat()
self.db.kv_set(kv_key, json.dumps(state))
client.add_comment(
task_id, f"❌ CheddahBot failed to complete this task.\n\nError: {str(e)[:2000]}"
)
# Move to "error" so Bryan can see what happened
client.update_task_status(task_id, self.config.clickup.error_status)
self._notify(
@ -554,7 +476,8 @@ class Scheduler:
age_ms = now_ms - updated_ms
if age_ms > threshold_ms:
reset_status = self.config.clickup.poll_statuses[0] if self.config.clickup.poll_statuses else "to do"
poll_sts = self.config.clickup.poll_statuses
reset_status = poll_sts[0] if poll_sts else "to do"
log.warning(
"Recovering stale task %s (%s) — stuck in '%s' for %.1f hours",
task.id, task.name, automation_status, age_ms / 3_600_000,
@ -572,22 +495,19 @@ class Scheduler:
category="clickup",
)
def _build_tool_args(self, state: dict) -> dict:
"""Build tool arguments from ClickUp task fields using the field mapping."""
skill_map = self.config.clickup.skill_map
task_type = state.get("task_type", "")
mapping = skill_map.get(task_type, {})
def _build_tool_args_from_task(self, task, mapping: dict) -> dict:
"""Build tool arguments from a ClickUp task using the field mapping."""
field_mapping = mapping.get("field_mapping", {})
args = {}
for tool_param, source in field_mapping.items():
if source == "task_name":
args[tool_param] = state.get("clickup_task_name", "")
args[tool_param] = task.name
elif source == "task_description":
args[tool_param] = state.get("custom_fields", {}).get("description", "")
args[tool_param] = task.custom_fields.get("description", "")
else:
# Look up custom field by name
args[tool_param] = state.get("custom_fields", {}).get(source, "")
args[tool_param] = task.custom_fields.get(source, "")
return args
@ -604,9 +524,7 @@ class Scheduler:
try:
self._auto_submit_cora_jobs()
self._poll_autocora_results()
self.db.kv_set(
"system:loop:autocora:last_run", datetime.now(UTC).isoformat()
)
self._loop_timestamps["autocora"] = datetime.now(UTC).isoformat()
except Exception as e:
log.error("AutoCora poll error: %s", e)
self._interruptible_wait(interval, self._force_autocora)
@ -707,9 +625,7 @@ class Scheduler:
while not self._stop_event.is_set():
try:
self._scan_watch_folder()
self.db.kv_set(
"system:loop:folder_watch:last_run", datetime.now(UTC).isoformat()
)
self._loop_timestamps["folder_watch"] = datetime.now(UTC).isoformat()
except Exception as e:
log.error("Folder watcher error: %s", e)
self._interruptible_wait(interval)
@ -726,30 +642,25 @@ class Scheduler:
log.debug("No .xlsx files in watch folder")
return
# Check processed/ subfolder for already-handled files
processed_dir = watch_folder / "processed"
processed_names = set()
if processed_dir.exists():
processed_names = {f.name for f in processed_dir.glob("*.xlsx")}
for xlsx_path in xlsx_files:
filename = xlsx_path.name
# Skip Office temp/lock files (e.g. ~$insert_molding.xlsx)
if filename.startswith("~$"):
continue
kv_key = f"linkbuilding:watched:{filename}"
# Skip completed/failed; retry "processing" (killed run) and "blocked" (missing field)
existing = self.db.kv_get(kv_key)
if existing:
try:
state = json.loads(existing)
if state.get("status") in ("completed", "failed"):
continue
if state.get("status") in ("processing", "blocked", "unmatched"):
log.info("Retrying '%s' state for %s", state["status"], filename)
self.db.kv_delete(kv_key)
except json.JSONDecodeError:
continue
# Skip files already in processed/
if filename in processed_names:
continue
log.info("Folder watcher: new .xlsx found: %s", filename)
self._process_watched_file(xlsx_path, kv_key)
self._process_watched_file(xlsx_path)
def _process_watched_file(self, xlsx_path: Path, kv_key: str):
def _process_watched_file(self, xlsx_path: Path):
"""Try to match a watched .xlsx file to a ClickUp task and run the pipeline."""
filename = xlsx_path.name
# Normalize filename stem for matching
@ -757,12 +668,6 @@ class Scheduler:
stem = xlsx_path.stem.lower().replace("-", " ").replace("_", " ")
stem = re.sub(r"\s+", " ", stem).strip()
# Mark as processing
self.db.kv_set(
kv_key,
json.dumps({"status": "processing", "started_at": datetime.now(UTC).isoformat()}),
)
# Try to find matching ClickUp task
matched_task = None
if self.config.clickup.enabled:
@ -770,17 +675,6 @@ class Scheduler:
if not matched_task:
log.warning("No ClickUp task match for '%s' — skipping", filename)
self.db.kv_set(
kv_key,
json.dumps(
{
"status": "unmatched",
"filename": filename,
"stem": stem,
"checked_at": datetime.now(UTC).isoformat(),
}
),
)
self._notify(
f"Folder watcher: no ClickUp match for **{filename}**.\n"
f"Create a Link Building task with Keyword "
@ -807,20 +701,6 @@ class Scheduler:
money_site_url = matched_task.custom_fields.get("IMSURL", "") or ""
if not money_site_url:
log.warning("Task %s (%s) missing IMSURL — skipping", task_id, matched_task.name)
self.db.kv_set(
kv_key,
json.dumps(
{
"status": "blocked",
"reason": "missing_imsurl",
"filename": filename,
"task_id": task_id,
"task_name": matched_task.name,
"checked_at": datetime.now(UTC).isoformat(),
}
),
)
# Set ClickUp status to "error" so it's visible on the board
client.update_task_status(task_id, self.config.clickup.error_status)
self._notify(
f"Folder watcher: **{filename}** matched task **{matched_task.name}** "
@ -853,20 +733,7 @@ class Scheduler:
result = "Error: tool registry not available"
if "Error" in result and "## Step" not in result:
# Pipeline failed
self.db.kv_set(
kv_key,
json.dumps(
{
"status": "failed",
"filename": filename,
"task_id": task_id,
"error": result[:500],
"failed_at": datetime.now(UTC).isoformat(),
}
),
)
client.update_task_status(task_id, self.config.clickup.error_status)
# Pipeline failed — tool handles its own ClickUp error status
self._notify(
f"Folder watcher: pipeline **failed** for **{filename}**.\n"
f"Error: {result[:200]}",
@ -883,18 +750,6 @@ class Scheduler:
except OSError as e:
log.warning("Could not move %s to processed: %s", filename, e)
self.db.kv_set(
kv_key,
json.dumps(
{
"status": "completed",
"filename": filename,
"task_id": task_id,
"completed_at": datetime.now(UTC).isoformat(),
}
),
)
client.update_task_status(task_id, "complete")
self._notify(
f"Folder watcher: pipeline **completed** for **{filename}**.\n"
f"ClickUp task: {matched_task.name}",
@ -903,18 +758,6 @@ class Scheduler:
except Exception as e:
log.error("Folder watcher pipeline error for %s: %s", filename, e)
self.db.kv_set(
kv_key,
json.dumps(
{
"status": "failed",
"filename": filename,
"task_id": task_id,
"error": str(e)[:500],
"failed_at": datetime.now(UTC).isoformat(),
}
),
)
client.update_task_status(task_id, self.config.clickup.error_status)
def _match_xlsx_to_clickup(self, normalized_stem: str):
@ -965,9 +808,7 @@ class Scheduler:
while not self._stop_event.is_set():
try:
self._scan_content_folder()
self.db.kv_set(
"system:loop:content_watch:last_run", datetime.now(UTC).isoformat()
)
self._loop_timestamps["content_watch"] = datetime.now(UTC).isoformat()
except Exception as e:
log.error("Content folder watcher error: %s", e)
self._interruptible_wait(interval)
@ -984,41 +825,30 @@ class Scheduler:
log.debug("No .xlsx files in content Cora inbox")
return
# Check processed/ subfolder for already-handled files
processed_dir = inbox / "processed"
processed_names = set()
if processed_dir.exists():
processed_names = {f.name for f in processed_dir.glob("*.xlsx")}
for xlsx_path in xlsx_files:
filename = xlsx_path.name
# Skip Office temp/lock files
if filename.startswith("~$"):
continue
kv_key = f"content:watched:{filename}"
# Skip completed/failed; retry processing/blocked/unmatched
existing = self.db.kv_get(kv_key)
if existing:
try:
state = json.loads(existing)
if state.get("status") in ("completed", "failed"):
continue
if state.get("status") in ("processing", "blocked", "unmatched"):
log.info("Retrying '%s' state for %s", state["status"], filename)
self.db.kv_delete(kv_key)
except json.JSONDecodeError:
continue
# Skip files already in processed/
if filename in processed_names:
continue
log.info("Content watcher: new .xlsx found: %s", filename)
self._process_content_file(xlsx_path, kv_key)
self._process_content_file(xlsx_path)
def _process_content_file(self, xlsx_path: Path, kv_key: str):
def _process_content_file(self, xlsx_path: Path):
"""Match a content Cora .xlsx to a ClickUp task and run create_content."""
filename = xlsx_path.name
stem = xlsx_path.stem.lower().replace("-", " ").replace("_", " ")
stem = re.sub(r"\s+", " ", stem).strip()
# Mark as processing
self.db.kv_set(
kv_key,
json.dumps({"status": "processing", "started_at": datetime.now(UTC).isoformat()}),
)
# Try to find matching ClickUp task
matched_task = None
if self.config.clickup.enabled:
@ -1026,17 +856,6 @@ class Scheduler:
if not matched_task:
log.warning("No ClickUp content task match for '%s' — skipping", filename)
self.db.kv_set(
kv_key,
json.dumps(
{
"status": "unmatched",
"filename": filename,
"stem": stem,
"checked_at": datetime.now(UTC).isoformat(),
}
),
)
self._notify(
f"Content watcher: no ClickUp match for **{filename}**.\n"
f"Create a Content Creation or On Page Optimization task with Keyword "
@ -1077,18 +896,6 @@ class Scheduler:
result = "Error: tool registry not available"
if result.startswith("Error:"):
self.db.kv_set(
kv_key,
json.dumps(
{
"status": "failed",
"filename": filename,
"task_id": task_id,
"error": result[:500],
"failed_at": datetime.now(UTC).isoformat(),
}
),
)
self._notify(
f"Content watcher: pipeline **failed** for **{filename}**.\n"
f"Error: {result[:200]}",
@ -1105,17 +912,6 @@ class Scheduler:
except OSError as e:
log.warning("Could not move %s to processed: %s", filename, e)
self.db.kv_set(
kv_key,
json.dumps(
{
"status": "completed",
"filename": filename,
"task_id": task_id,
"completed_at": datetime.now(UTC).isoformat(),
}
),
)
self._notify(
f"Content watcher: pipeline **completed** for **{filename}**.\n"
f"ClickUp task: {matched_task.name}",
@ -1124,18 +920,6 @@ class Scheduler:
except Exception as e:
log.error("Content watcher pipeline error for %s: %s", filename, e)
self.db.kv_set(
kv_key,
json.dumps(
{
"status": "failed",
"filename": filename,
"task_id": task_id,
"error": str(e)[:500],
"failed_at": datetime.now(UTC).isoformat(),
}
),
)
def _match_xlsx_to_content_task(self, normalized_stem: str):
"""Find a ClickUp content task whose Keyword matches the file stem.

175
cheddahbot/tools/clickup_tool.py
View File

@ -1,8 +1,7 @@
"""ClickUp chat-facing tools for listing, approving, and declining tasks."""
"""ClickUp chat-facing tools for listing, querying, and resetting tasks."""
from __future__ import annotations
import json
import logging
from . import tool
@ -24,22 +23,6 @@ def _get_clickup_client(ctx: dict):
)
def _get_clickup_states(db) -> dict[str, dict]:
"""Load all tracked ClickUp task states from kv_store."""
pairs = db.kv_scan("clickup:task:")
states = {}
for key, value in pairs:
# keys look like clickup:task:{id}:state
parts = key.split(":")
if len(parts) == 4 and parts[3] == "state":
task_id = parts[2]
try: # noqa: SIM105
states[task_id] = json.loads(value)
except json.JSONDecodeError:
pass
return states
@tool(
"clickup_query_tasks",
"Query ClickUp live for tasks. Optionally filter by status (e.g. 'to do', 'in progress') "
@ -94,112 +77,116 @@ def clickup_query_tasks(status: str = "", task_type: str = "", ctx: dict | None
@tool(
"clickup_list_tasks",
"List ClickUp tasks that Cheddah is tracking. Optionally filter by internal state "
"(executing, completed, failed).",
"List ClickUp tasks in automation-related statuses (automation underway, "
"outline review, internal review, error). Shows tasks currently being processed.",
category="clickup",
)
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)
"""List ClickUp tasks in automation-related statuses."""
client = _get_clickup_client(ctx)
if not client:
return "Error: ClickUp API token not configured."
if not states:
return "No ClickUp tasks are currently being tracked."
cfg = ctx["config"].clickup
if not cfg.space_id:
return "Error: ClickUp space_id not configured."
# Query tasks in automation-related statuses
automation_statuses = [
cfg.automation_status,
"outline review",
cfg.review_status,
cfg.error_status,
]
if status:
states = {tid: s for tid, s in states.items() if s.get("state") == status}
if not states:
return f"No ClickUp tasks with state '{status}'."
automation_statuses = [status]
try:
tasks = client.get_tasks_from_space(cfg.space_id, statuses=automation_statuses)
except Exception as e:
return f"Error querying ClickUp: {e}"
finally:
client.close()
if not tasks:
filter_note = f" with status '{status}'" if status else " in automation statuses"
return f"No tasks found{filter_note}."
lines = []
for task_id, state in sorted(states.items(), key=lambda x: x[1].get("discovered_at", "")):
name = state.get("clickup_task_name", "Unknown")
task_type = state.get("task_type", "—")
task_state = state.get("state", "unknown")
skill = state.get("skill_name", "—")
lines.append(
f"• **{name}** (ID: {task_id})\n"
f" Type: {task_type} | State: {task_state} | Skill: {skill}"
)
for t in tasks:
parts = [f"**{t.name}** (ID: {t.id})"]
parts.append(f" Status: {t.status} | Type: {t.task_type or '—'}")
fields = {k: v for k, v in t.custom_fields.items() if v}
if fields:
field_strs = [f"{k}: {v}" for k, v in fields.items()]
parts.append(f" Fields: {', '.join(field_strs)}")
lines.append("\n".join(parts))
return f"**Tracked ClickUp Tasks ({len(lines)}):**\n\n" + "\n\n".join(lines)
return f"**Automation Tasks ({len(lines)}):**\n\n" + "\n\n".join(lines)
@tool(
"clickup_task_status",
"Check the detailed internal processing state of a ClickUp task by its ID.",
"Check the current status and details of a ClickUp task by its ID.",
category="clickup",
)
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")
if not raw:
return f"No tracked state found for task ID '{task_id}'."
"""Get current status for a specific ClickUp task from the API."""
client = _get_clickup_client(ctx)
if not client:
return "Error: ClickUp API token not configured."
try:
state = json.loads(raw)
except json.JSONDecodeError:
return f"Corrupted state data for task '{task_id}'."
task = client.get_task(task_id)
except Exception as e:
return f"Error fetching task '{task_id}': {e}"
finally:
client.close()
lines = [f"**Task: {state.get('clickup_task_name', 'Unknown')}** (ID: {task_id})"]
lines.append(f"State: {state.get('state', 'unknown')}")
lines.append(f"Task Type: {state.get('task_type', '—')}")
lines.append(f"Mapped Skill: {state.get('skill_name', '—')}")
lines.append(f"Discovered: {state.get('discovered_at', '—')}")
if state.get("started_at"):
lines.append(f"Started: {state['started_at']}")
if state.get("completed_at"):
lines.append(f"Completed: {state['completed_at']}")
if state.get("error"):
lines.append(f"Error: {state['error']}")
if state.get("deliverable_paths"):
lines.append(f"Deliverables: {', '.join(state['deliverable_paths'])}")
if state.get("custom_fields"):
fields_str = ", ".join(f"{k}: {v}" for k, v in state["custom_fields"].items() if v)
if fields_str:
lines.append(f"Custom Fields: {fields_str}")
lines = [f"**Task: {task.name}** (ID: {task.id})"]
lines.append(f"Status: {task.status}")
lines.append(f"Type: {task.task_type or '—'}")
if task.url:
lines.append(f"URL: {task.url}")
if task.due_date:
lines.append(f"Due: {task.due_date}")
if task.date_updated:
lines.append(f"Updated: {task.date_updated}")
fields = {k: v for k, v in task.custom_fields.items() if v}
if fields:
field_strs = [f"{k}: {v}" for k, v in fields.items()]
lines.append(f"Fields: {', '.join(field_strs)}")
return "\n".join(lines)
@tool(
"clickup_reset_task",
"Reset a ClickUp task's internal tracking state so it can be retried on the next poll. "
"Use this when a task has failed or completed and you want to re-run it.",
"Reset a ClickUp task to 'to do' status so it can be retried on the next poll. "
"Use this when a task is stuck in an error or automation state.",
category="clickup",
)
def clickup_reset_task(task_id: str, ctx: dict | None = None) -> str:
"""Delete the kv_store state for a single task so it can be retried."""
db = ctx["db"]
key = f"clickup:task:{task_id}:state"
raw = db.kv_get(key)
if not raw:
return f"No tracked state found for task ID '{task_id}'. Nothing to reset."
"""Reset a ClickUp task status to 'to do' for retry."""
client = _get_clickup_client(ctx)
if not client:
return "Error: ClickUp API token not configured."
db.kv_delete(key)
return f"Task '{task_id}' state cleared. It will be picked up on the next scheduler poll."
cfg = ctx["config"].clickup
reset_status = cfg.poll_statuses[0] if cfg.poll_statuses else "to do"
@tool(
"clickup_reset_all",
"Clear ALL internal ClickUp task tracking state. Use this to wipe the slate clean "
"so all eligible tasks can be retried on the next poll cycle.",
category="clickup",
)
def clickup_reset_all(ctx: dict | None = None) -> str:
"""Delete all clickup task states and legacy active_ids from kv_store."""
db = ctx["db"]
states = _get_clickup_states(db)
count = 0
for task_id in states:
db.kv_delete(f"clickup:task:{task_id}:state")
count += 1
# Also clean up legacy active_ids key
if db.kv_get("clickup:active_task_ids"):
db.kv_delete("clickup:active_task_ids")
try:
client.update_task_status(task_id, reset_status)
client.add_comment(
task_id,
f"Task reset to '{reset_status}' via chat command.",
)
except Exception as e:
return f"Error resetting task '{task_id}': {e}"
finally:
client.close()
return (
f"Cleared {count} task state(s) from tracking. Next poll will re-discover eligible tasks."
f"Task '{task_id}' reset to '{reset_status}'. "
f"It will be picked up on the next scheduler poll."
)

176
cheddahbot/tools/content_creation.py
View File

@ -9,10 +9,8 @@ The content-researcher skill in the execution brain is triggered by keywords lik
from __future__ import annotations
import json
import logging
import re
from datetime import UTC, datetime
from pathlib import Path
from . import tool
@ -361,7 +359,8 @@ def _build_phase2_prompt(
@tool(
"create_content",
"Two-phase SEO content creation: Phase 1 researches + outlines, Phase 2 writes "
"full content from the approved outline. Auto-detects phase from kv_store state. "
"full content from the approved outline. Auto-detects phase from ClickUp task "
"status ('outline approved' → Phase 2). "
"Auto-detects content type from URL presence if not specified.",
category="content",
)
@ -396,20 +395,20 @@ def create_content(
config = ctx.get("config")
db = ctx.get("db")
task_id = ctx.get("clickup_task_id", "")
kv_key = f"clickup:task:{task_id}:state" if task_id else ""
# Determine phase from kv_store state
# Determine phase from ClickUp task status
phase = 1
existing_state = {}
if kv_key and db:
raw = db.kv_get(kv_key)
if raw:
if task_id and ctx:
client = _get_clickup_client(ctx)
if client:
try:
existing_state = json.loads(raw)
if existing_state.get("state") == "outline_review":
task = client.get_task(task_id)
if task.status.lower() == "outline approved":
phase = 2
except json.JSONDecodeError:
pass
except Exception as e:
log.warning("Could not check ClickUp status for phase detection: %s", e)
finally:
client.close()
# Find Cora report
cora_inbox = config.content.cora_inbox if config else ""
@ -426,7 +425,6 @@ def create_content(
db=db,
ctx=ctx,
task_id=task_id,
kv_key=kv_key,
url=url,
keyword=keyword,
content_type=content_type,
@ -441,11 +439,10 @@ def create_content(
db=db,
ctx=ctx,
task_id=task_id,
kv_key=kv_key,
url=url,
keyword=keyword,
cora_path=cora_path,
existing_state=existing_state,
existing_state={},
is_service_page=is_service_page,
capabilities_default=capabilities_default,
)
@ -463,7 +460,6 @@ def _run_phase1(
db,
ctx,
task_id: str,
kv_key: str,
url: str,
keyword: str,
content_type: str,
@ -471,8 +467,6 @@ def _run_phase1(
capabilities_default: str,
is_service_page: bool = False,
) -> str:
now = datetime.now(UTC).isoformat()
# ClickUp: move to automation underway
if task_id:
_sync_clickup_start(ctx, task_id)
@ -492,13 +486,11 @@ def _run_phase1(
error_msg = f"Phase 1 execution failed: {e}"
log.error(error_msg)
if task_id:
_update_kv_state(db, kv_key, "failed", error=str(e))
_sync_clickup_fail(ctx, task_id, str(e))
return f"Error: {error_msg}"
if result.startswith("Error:"):
if task_id:
_update_kv_state(db, kv_key, "failed", error=result)
_sync_clickup_fail(ctx, task_id, result)
return result
@ -506,23 +498,7 @@ def _run_phase1(
outline_path = _save_content(result, keyword, "outline.md", config)
log.info("Outline saved to: %s", outline_path)
# Update kv_store
if kv_key and db:
state = {
"state": "outline_review",
"clickup_task_id": task_id,
"url": url,
"keyword": keyword,
"content_type": content_type,
"cora_path": cora_path,
"outline_path": outline_path,
"phase1_completed_at": now,
"completed_at": None,
"error": None,
}
db.kv_set(kv_key, json.dumps(state))
# ClickUp: move to outline review
# ClickUp: move to outline review + store OutlinePath
if task_id:
_sync_clickup_outline_ready(ctx, task_id, outline_path)
@ -585,7 +561,6 @@ def _run_phase2(
db,
ctx,
task_id: str,
kv_key: str,
url: str,
keyword: str,
cora_path: str,
@ -593,11 +568,8 @@ def _run_phase2(
is_service_page: bool = False,
capabilities_default: str = "",
) -> str:
# Resolve outline path: ClickUp field → convention → state fallback
# Resolve outline path: ClickUp field → convention
outline_path = _resolve_outline_path(ctx, task_id, keyword, config)
if not outline_path:
# Last resort: check existing_state (for continue_content calls)
outline_path = existing_state.get("outline_path", "")
outline_text = ""
if outline_path:
@ -612,7 +584,8 @@ def _run_phase2(
client = _get_clickup_client(ctx)
if client:
try:
reset_status = config.clickup.poll_statuses[0] if config.clickup.poll_statuses else "to do"
poll_sts = config.clickup.poll_statuses
reset_status = poll_sts[0] if poll_sts else "to do"
client.update_task_status(task_id, reset_status)
client.add_comment(
task_id,
@ -653,13 +626,11 @@ def _run_phase2(
error_msg = f"Phase 2 execution failed: {e}"
log.error(error_msg)
if task_id:
_update_kv_state(db, kv_key, "failed", error=str(e))
_sync_clickup_fail(ctx, task_id, str(e))
return f"Error: {error_msg}"
if result.startswith("Error:"):
if task_id:
_update_kv_state(db, kv_key, "failed", error=result)
_sync_clickup_fail(ctx, task_id, result)
return result
@ -667,16 +638,6 @@ def _run_phase2(
content_path = _save_content(result, keyword, "final-content.md", config)
log.info("Final content saved to: %s", content_path)
# Update kv_store
if kv_key and db:
now = datetime.now(UTC).isoformat()
state = existing_state.copy()
state["state"] = "completed"
state["content_path"] = content_path
state["completed_at"] = now
state["error"] = None
db.kv_set(kv_key, json.dumps(state))
# ClickUp: move to internal review
if task_id:
_sync_clickup_complete(ctx, task_id, content_path)
@ -714,65 +675,64 @@ def continue_content(
"""
if not keyword:
return "Error: 'keyword' is required."
if not ctx or "agent" not in ctx or "db" not in ctx:
return "Error: Tool context with agent and db is required."
if not ctx or "agent" not in ctx:
return "Error: Tool context with agent is required."
db = ctx["db"]
config = ctx.get("config")
db = ctx.get("db")
# Scan kv_store for outline_review entries matching keyword
entries = db.kv_scan("clickup:task:")
keyword_lower = keyword.lower().strip()
for key, raw in entries:
# Query ClickUp for tasks in "outline approved" or "outline review" status
# matching the keyword
client = _get_clickup_client(ctx)
if client:
try:
state = json.loads(raw)
except (json.JSONDecodeError, TypeError):
continue
if state.get("state") != "outline_review":
continue
if state.get("keyword", "").lower().strip() == keyword_lower:
# Found a matching entry — run Phase 2
task_id = state.get("clickup_task_id", "")
kv_key = key
url = state.get("url", "")
cora_path = state.get("cora_path", "")
space_id = config.clickup.space_id if config else ""
if space_id:
tasks = client.get_tasks_from_space(
space_id,
statuses=["outline approved", "outline review"],
)
keyword_lower = keyword.lower().strip()
for task in tasks:
task_keyword = task.custom_fields.get("Keyword", "")
if str(task_keyword).lower().strip() == keyword_lower:
task_id = task.id
url = task.custom_fields.get("IMSURL", "") or ""
cora_inbox = config.content.cora_inbox if config else ""
cora_path = _find_cora_report(keyword, cora_inbox)
return _run_phase2(
agent=ctx["agent"],
config=config,
db=db,
ctx=ctx,
task_id=task_id,
kv_key=kv_key,
url=url,
keyword=keyword,
cora_path=cora_path,
existing_state=state,
)
return _run_phase2(
agent=ctx["agent"],
config=config,
db=db,
ctx=ctx,
task_id=task_id,
url=str(url),
keyword=keyword,
cora_path=cora_path or "",
existing_state={},
)
except Exception as e:
log.warning("ClickUp query failed in continue_content: %s", e)
finally:
client.close()
# Fallback: try to run Phase 2 without a ClickUp task (outline must exist locally)
outline_path = _resolve_outline_path(ctx, "", keyword, config)
if outline_path:
return _run_phase2(
agent=ctx["agent"],
config=config,
db=db,
ctx=ctx,
task_id="",
url="",
keyword=keyword,
cora_path="",
existing_state={},
)
return (
f"No outline awaiting review found for keyword '{keyword}'. "
f"Use create_content to start Phase 1 first."
)
# ---------------------------------------------------------------------------
# KV state helper
# ---------------------------------------------------------------------------
def _update_kv_state(db, kv_key: str, state_val: str, error: str = "") -> None:
"""Update kv_store state without losing existing data."""
if not db or not kv_key:
return
raw = db.kv_get(kv_key)
try:
state = json.loads(raw) if raw else {}
except json.JSONDecodeError:
state = {}
state["state"] = state_val
if error:
state["error"] = error[:2000]
state["completed_at"] = datetime.now(UTC).isoformat()
db.kv_set(kv_key, json.dumps(state))

95
cheddahbot/tools/linkbuilding.py
View File

@ -6,12 +6,10 @@ Primary workflow: ingest CORA .xlsx → generate content batch.
from __future__ import annotations
import json
import logging
import os
import re
import subprocess
from datetime import UTC, datetime
from pathlib import Path
from . import tool
@ -163,9 +161,9 @@ def _parse_generate_output(stdout: str) -> dict:
def _set_status(ctx: dict | None, message: str) -> None:
"""Write pipeline progress to KV store for UI polling."""
if ctx and "db" in ctx:
ctx["db"].kv_set("linkbuilding:status", message)
"""Log pipeline progress. Previously wrote to KV; now just logs."""
if message:
log.info("[LB Pipeline] %s", message)
def _get_clickup_client(ctx: dict | None):
@ -187,25 +185,10 @@ def _get_clickup_client(ctx: dict | None):
def _sync_clickup(ctx: dict | None, task_id: str, step: str, message: str) -> None:
"""Post a comment to ClickUp and update KV state."""
"""Post a progress comment to ClickUp."""
if not task_id or not ctx:
return
# Update KV store
db = ctx.get("db")
if db:
kv_key = f"clickup:task:{task_id}:state"
raw = db.kv_get(kv_key)
if raw:
try:
state = json.loads(raw)
state["last_step"] = step
state["last_message"] = message
db.kv_set(kv_key, json.dumps(state))
except json.JSONDecodeError:
pass
# Post comment to ClickUp
cu_client = _get_clickup_client(ctx)
if cu_client:
try:
@ -254,26 +237,8 @@ def _find_clickup_task(ctx: dict, keyword: str) -> str:
continue
if _fuzzy_keyword_match(keyword_norm, _normalize_for_match(str(task_keyword))):
# Found a match — create executing state
# Found a match — move to "automation underway"
task_id = task.id
now = datetime.now(UTC).isoformat()
state = {
"state": "executing",
"clickup_task_id": task_id,
"clickup_task_name": task.name,
"task_type": task.task_type,
"skill_name": "run_link_building",
"discovered_at": now,
"started_at": now,
"completed_at": None,
"error": None,
"deliverable_paths": [],
"custom_fields": task.custom_fields,
}
db = ctx.get("db")
if db:
db.kv_set(f"clickup:task:{task_id}:state", json.dumps(state))
# Move to "automation underway"
cu_client2 = _get_clickup_client(ctx)
@ -322,7 +287,7 @@ def _fuzzy_keyword_match(a: str, b: str) -> bool:
def _complete_clickup_task(ctx: dict | None, task_id: str, message: str, status: str = "") -> None:
"""Mark a ClickUp task as completed and update KV state."""
"""Mark a ClickUp task as completed."""
if not task_id or not ctx:
return
@ -331,19 +296,6 @@ def _complete_clickup_task(ctx: dict | None, task_id: str, message: str, status:
lb_map = skill_map.get("Link Building", {})
complete_status = status or lb_map.get("complete_status", "complete")
db = ctx.get("db")
if db:
kv_key = f"clickup:task:{task_id}:state"
raw = db.kv_get(kv_key)
if raw:
try:
state = json.loads(raw)
state["state"] = "completed"
state["completed_at"] = datetime.now(UTC).isoformat()
db.kv_set(kv_key, json.dumps(state))
except json.JSONDecodeError:
pass
cu_client = _get_clickup_client(ctx)
if cu_client:
try:
@ -356,27 +308,13 @@ def _complete_clickup_task(ctx: dict | None, task_id: str, message: str, status:
def _fail_clickup_task(ctx: dict | None, task_id: str, error_msg: str) -> None:
"""Mark a ClickUp task as failed and update KV state."""
"""Mark a ClickUp task as failed."""
if not task_id or not ctx:
return
config = ctx.get("config")
error_status = config.clickup.error_status if config else "error"
db = ctx.get("db")
if db:
kv_key = f"clickup:task:{task_id}:state"
raw = db.kv_get(kv_key)
if raw:
try:
state = json.loads(raw)
state["state"] = "failed"
state["error"] = error_msg
state["completed_at"] = datetime.now(UTC).isoformat()
db.kv_set(kv_key, json.dumps(state))
except json.JSONDecodeError:
pass
cu_client = _get_clickup_client(ctx)
if cu_client:
try:
@ -749,7 +687,6 @@ def scan_cora_folder(ctx: dict | None = None) -> str:
if not watch_path.exists():
return f"Watch folder does not exist: {watch_folder}"
db = ctx.get("db")
xlsx_files = sorted(watch_path.glob("*.xlsx"))
if not xlsx_files:
@ -757,18 +694,16 @@ def scan_cora_folder(ctx: dict | None = None) -> str:
lines = [f"## Cora Inbox: {watch_folder}\n"]
processed_dir = watch_path / "processed"
processed_names = set()
if processed_dir.exists():
processed_names = {f.name for f in processed_dir.glob("*.xlsx")}
for f in xlsx_files:
filename = f.name
status = "new"
if db:
kv_val = db.kv_get(f"linkbuilding:watched:{filename}")
if kv_val:
try:
watched = json.loads(kv_val)
status = watched.get("status", "unknown")
except json.JSONDecodeError:
status = "tracked"
if filename.startswith("~$"):
continue
status = "processed" if filename in processed_names else "new"
lines.append(f"- **{filename}** — status: {status}")
# Check processed subfolder

40
cheddahbot/tools/press_release.py
View File

@ -14,7 +14,7 @@ import json
import logging
import re
import time
from datetime import UTC, datetime
from datetime import datetime
from pathlib import Path
from ..docx_export import text_to_docx
@ -38,9 +38,9 @@ SONNET_CLI_MODEL = "sonnet"
def _set_status(ctx: dict | None, message: str) -> None:
"""Write pipeline progress to the DB so the UI can poll it."""
if ctx and "db" in ctx:
ctx["db"].kv_set("pipeline:status", message)
"""Log pipeline progress. Previously wrote to KV; now just logs."""
if message:
log.info("[PR Pipeline] %s", message)
def _fuzzy_company_match(name: str, candidate: str) -> bool:
@ -95,26 +95,8 @@ def _find_clickup_task(ctx: dict, company_name: str) -> str:
):
continue
# Found a match — create kv_store entry and move to "in progress"
# Found a match — move to "automation underway" on ClickUp
task_id = task.id
now = datetime.now(UTC).isoformat()
state = {
"state": "executing",
"clickup_task_id": task_id,
"clickup_task_name": task.name,
"task_type": task.task_type,
"skill_name": "write_press_releases",
"discovered_at": now,
"started_at": now,
"completed_at": None,
"error": None,
"deliverable_paths": [],
"custom_fields": task.custom_fields,
}
db = ctx.get("db")
if db:
db.kv_set(f"clickup:task:{task_id}:state", json.dumps(state))
# Move to "automation underway" on ClickUp
cu_client2 = _get_clickup_client(ctx)
@ -809,18 +791,6 @@ def write_press_releases(
# Set status to internal review
cu_client.update_task_status(clickup_task_id, config.clickup.review_status)
# Update kv_store state if one exists
db = ctx.get("db")
if db:
kv_key = f"clickup:task:{clickup_task_id}:state"
existing = db.kv_get(kv_key)
if existing:
state = json.loads(existing)
state["state"] = "completed"
state["completed_at"] = datetime.now(UTC).isoformat()
state["deliverable_paths"] = docx_files
db.kv_set(kv_key, json.dumps(state))
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")

8
cheddahbot/ui.py
View File

@ -454,13 +454,7 @@ def create_ui(
return agent_name, agent_name, chatbot_msgs, convs, new_browser
def poll_pipeline_status(agent_name):
"""Poll the DB for pipeline progress updates."""
agent = _get_agent(agent_name)
if not agent:
return gr.update(value="", visible=False)
status = agent.db.kv_get("pipeline:status")
if status:
return gr.update(value=f"⏳ {status}", visible=True)
"""Pipeline status indicator (no longer used — kept for UI timer)."""
return gr.update(value="", visible=False)
def poll_notifications():

223
tests/test_clickup_tools.py
View File

@ -1,147 +1,144 @@
"""Tests for the ClickUp chat tools."""
"""Tests for the ClickUp chat tools (API-backed, no KV store)."""
from __future__ import annotations
import json
from dataclasses import dataclass, field
from unittest.mock import MagicMock, patch
from cheddahbot.tools.clickup_tool import (
clickup_list_tasks,
clickup_reset_all,
clickup_query_tasks,
clickup_reset_task,
clickup_task_status,
)
def _make_ctx(db):
return {"db": db}
@dataclass
class FakeTask:
id: str = "t1"
name: str = "Test Task"
status: str = "to do"
task_type: str = "Press Release"
url: str = "https://app.clickup.com/t/t1"
due_date: str = ""
date_updated: str = ""
tags: list = field(default_factory=list)
custom_fields: dict = field(default_factory=dict)
def _seed_task(db, task_id, state, **overrides):
"""Insert a task state into kv_store."""
data = {
"state": state,
"clickup_task_id": task_id,
"clickup_task_name": f"Task {task_id}",
"task_type": "Press Release",
"skill_name": "write_press_releases",
"discovered_at": "2026-01-01T00:00:00",
"started_at": None,
"completed_at": None,
"error": None,
"deliverable_paths": [],
"custom_fields": {},
}
data.update(overrides)
db.kv_set(f"clickup:task:{task_id}:state", json.dumps(data))
def _make_ctx():
config = MagicMock()
config.clickup.api_token = "test-token"
config.clickup.workspace_id = "ws1"
config.clickup.space_id = "sp1"
config.clickup.task_type_field_name = "Work Category"
config.clickup.automation_status = "automation underway"
config.clickup.review_status = "internal review"
config.clickup.error_status = "error"
config.clickup.poll_statuses = ["to do"]
return {"config": config, "db": MagicMock()}
class TestClickupQueryTasks:
@patch("cheddahbot.tools.clickup_tool._get_clickup_client")
def test_returns_tasks(self, mock_client_fn):
mock_client = MagicMock()
mock_client.get_tasks_from_space.return_value = [
FakeTask(id="t1", name="PR Task", task_type="Press Release"),
]
mock_client_fn.return_value = mock_client
result = clickup_query_tasks(ctx=_make_ctx())
assert "PR Task" in result
assert "t1" in result
@patch("cheddahbot.tools.clickup_tool._get_clickup_client")
def test_no_tasks_found(self, mock_client_fn):
mock_client = MagicMock()
mock_client.get_tasks_from_space.return_value = []
mock_client_fn.return_value = mock_client
result = clickup_query_tasks(ctx=_make_ctx())
assert "No tasks found" in result
class TestClickupListTasks:
def test_empty_when_no_tasks(self, tmp_db):
result = clickup_list_tasks(ctx=_make_ctx(tmp_db))
assert "No ClickUp tasks" in result
@patch("cheddahbot.tools.clickup_tool._get_clickup_client")
def test_lists_automation_tasks(self, mock_client_fn):
mock_client = MagicMock()
mock_client.get_tasks_from_space.return_value = [
FakeTask(id="t1", name="Active Task", status="automation underway"),
]
mock_client_fn.return_value = mock_client
def test_lists_all_tracked_tasks(self, tmp_db):
_seed_task(tmp_db, "a1", "discovered")
_seed_task(tmp_db, "a2", "approved")
result = clickup_list_tasks(ctx=_make_ctx())
assert "Active Task" in result
assert "t1" in result
result = clickup_list_tasks(ctx=_make_ctx(tmp_db))
@patch("cheddahbot.tools.clickup_tool._get_clickup_client")
def test_no_automation_tasks(self, mock_client_fn):
mock_client = MagicMock()
mock_client.get_tasks_from_space.return_value = []
mock_client_fn.return_value = mock_client
assert "a1" in result
assert "a2" in result
assert "2" in result # count
result = clickup_list_tasks(ctx=_make_ctx())
assert "No tasks found" in result
def test_filter_by_status(self, tmp_db):
_seed_task(tmp_db, "a1", "discovered")
_seed_task(tmp_db, "a2", "approved")
_seed_task(tmp_db, "a3", "completed")
@patch("cheddahbot.tools.clickup_tool._get_clickup_client")
def test_filter_by_status(self, mock_client_fn):
mock_client = MagicMock()
mock_client.get_tasks_from_space.return_value = [
FakeTask(id="t1", name="Error Task", status="error"),
]
mock_client_fn.return_value = mock_client
result = clickup_list_tasks(status="approved", ctx=_make_ctx(tmp_db))
assert "a2" in result
assert "a1" not in result
assert "a3" not in result
def test_filter_returns_empty_message(self, tmp_db):
_seed_task(tmp_db, "a1", "discovered")
result = clickup_list_tasks(status="completed", ctx=_make_ctx(tmp_db))
assert "No ClickUp tasks with state" in result
result = clickup_list_tasks(status="error", ctx=_make_ctx())
assert "Error Task" in result
class TestClickupTaskStatus:
def test_shows_details(self, tmp_db):
_seed_task(tmp_db, "a1", "executing", started_at="2026-01-01T12:00:00")
@patch("cheddahbot.tools.clickup_tool._get_clickup_client")
def test_shows_details(self, mock_client_fn):
mock_client = MagicMock()
mock_client.get_task.return_value = FakeTask(
id="t1",
name="My Task",
status="automation underway",
task_type="Press Release",
)
mock_client_fn.return_value = mock_client
result = clickup_task_status(task_id="a1", ctx=_make_ctx(tmp_db))
assert "Task a1" in result
assert "executing" in result
result = clickup_task_status(task_id="t1", ctx=_make_ctx())
assert "My Task" in result
assert "automation underway" in result
assert "Press Release" in result
assert "2026-01-01T12:00:00" in result
def test_unknown_task(self, tmp_db):
result = clickup_task_status(task_id="nonexistent", ctx=_make_ctx(tmp_db))
@patch("cheddahbot.tools.clickup_tool._get_clickup_client")
def test_api_error(self, mock_client_fn):
mock_client = MagicMock()
mock_client.get_task.side_effect = Exception("Not found")
mock_client_fn.return_value = mock_client
assert "No tracked state" in result
def test_shows_error_when_failed(self, tmp_db):
_seed_task(tmp_db, "f1", "failed", error="API timeout")
result = clickup_task_status(task_id="f1", ctx=_make_ctx(tmp_db))
assert "API timeout" in result
def test_shows_deliverables(self, tmp_db):
_seed_task(tmp_db, "c1", "completed", deliverable_paths=["/data/pr1.txt", "/data/pr2.txt"])
result = clickup_task_status(task_id="c1", ctx=_make_ctx(tmp_db))
assert "/data/pr1.txt" in result
result = clickup_task_status(task_id="bad", ctx=_make_ctx())
assert "Error" in result
class TestClickupResetTask:
def test_resets_failed_task(self, tmp_db):
_seed_task(tmp_db, "f1", "failed")
@patch("cheddahbot.tools.clickup_tool._get_clickup_client")
def test_resets_task(self, mock_client_fn):
mock_client = MagicMock()
mock_client_fn.return_value = mock_client
result = clickup_reset_task(task_id="f1", ctx=_make_ctx(tmp_db))
result = clickup_reset_task(task_id="t1", ctx=_make_ctx())
assert "reset" in result.lower()
mock_client.update_task_status.assert_called_once_with("t1", "to do")
mock_client.add_comment.assert_called_once()
assert "cleared" in result.lower()
assert tmp_db.kv_get("clickup:task:f1:state") is None
@patch("cheddahbot.tools.clickup_tool._get_clickup_client")
def test_api_error(self, mock_client_fn):
mock_client = MagicMock()
mock_client.update_task_status.side_effect = Exception("API error")
mock_client_fn.return_value = mock_client
def test_resets_completed_task(self, tmp_db):
_seed_task(tmp_db, "c1", "completed")
result = clickup_reset_task(task_id="c1", ctx=_make_ctx(tmp_db))
assert "cleared" in result.lower()
assert tmp_db.kv_get("clickup:task:c1:state") is None
def test_unknown_task(self, tmp_db):
result = clickup_reset_task(task_id="nope", ctx=_make_ctx(tmp_db))
assert "Nothing to reset" in result
class TestClickupResetAll:
def test_clears_all_states(self, tmp_db):
_seed_task(tmp_db, "a1", "completed")
_seed_task(tmp_db, "a2", "failed")
_seed_task(tmp_db, "a3", "executing")
result = clickup_reset_all(ctx=_make_ctx(tmp_db))
assert "3" in result
assert tmp_db.kv_get("clickup:task:a1:state") is None
assert tmp_db.kv_get("clickup:task:a2:state") is None
assert tmp_db.kv_get("clickup:task:a3:state") is None
def test_clears_legacy_active_ids(self, tmp_db):
tmp_db.kv_set("clickup:active_task_ids", json.dumps(["a1", "a2"]))
clickup_reset_all(ctx=_make_ctx(tmp_db))
assert tmp_db.kv_get("clickup:active_task_ids") is None
def test_empty_returns_zero(self, tmp_db):
result = clickup_reset_all(ctx=_make_ctx(tmp_db))
assert "0" in result
result = clickup_reset_task(task_id="t1", ctx=_make_ctx())
assert "Error" in result

133
tests/test_content_creation.py
View File

@ -2,7 +2,6 @@
from __future__ import annotations
import json
from pathlib import Path
from unittest.mock import MagicMock, patch
@ -261,20 +260,22 @@ class TestCreateContentPhase1:
saved = (outline_dir / "outline.md").read_text(encoding="utf-8")
assert saved == "## Generated Outline\nSection 1..."
def test_phase1_sets_kv_state(self, tmp_db, tmp_path):
@patch("cheddahbot.tools.content_creation._get_clickup_client")
def test_phase1_syncs_clickup(self, mock_get_client, tmp_db, tmp_path):
mock_client = MagicMock()
mock_get_client.return_value = mock_client
ctx = self._make_ctx(tmp_db, tmp_path)
create_content(
url="https://example.com",
keyword="plumbing services",
ctx=ctx,
)
raw = tmp_db.kv_get("clickup:task:task123:state")
assert raw is not None
state = json.loads(raw)
assert state["state"] == "outline_review"
assert state["keyword"] == "plumbing services"
assert state["url"] == "https://example.com"
assert "outline_path" in state
# Verify outline review status was set and OutlinePath was stored
mock_client.update_task_status.assert_any_call("task123", "outline review")
mock_client.set_custom_field_by_name.assert_called_once()
call_args = mock_client.set_custom_field_by_name.call_args
assert call_args[0][0] == "task123"
assert call_args[0][1] == "OutlinePath"
def test_phase1_includes_clickup_sync_marker(self, tmp_db, tmp_path):
ctx = self._make_ctx(tmp_db, tmp_path)
@ -293,7 +294,7 @@ class TestCreateContentPhase1:
class TestCreateContentPhase2:
def _setup_phase2(self, tmp_db, tmp_path):
"""Set up an outline_review state and outline file, return ctx."""
"""Set up outline file and return (ctx, outline_path)."""
cfg = Config()
cfg.content = ContentConfig(outline_dir=str(tmp_path / "outlines"))
@ -303,29 +304,30 @@ class TestCreateContentPhase2:
outline_file = outline_dir / "outline.md"
outline_file.write_text("## Approved Outline\nSection content here.", encoding="utf-8")
# Set kv_store to outline_review
state = {
"state": "outline_review",
"clickup_task_id": "task456",
"url": "https://example.com/plumbing",
"keyword": "plumbing services",
"content_type": "service page",
"cora_path": "",
"outline_path": str(outline_file),
}
tmp_db.kv_set("clickup:task:task456:state", json.dumps(state))
agent = MagicMock()
agent.execute_task.return_value = "# Full Content\nParagraph..."
return {
ctx = {
"agent": agent,
"config": cfg,
"db": tmp_db,
"clickup_task_id": "task456",
}
return ctx, str(outline_file)
def _make_phase2_client(self, outline_path):
"""Create a mock ClickUp client that triggers Phase 2 detection."""
mock_client = MagicMock()
mock_task = MagicMock()
mock_task.status = "outline approved"
mock_client.get_task.return_value = mock_task
mock_client.get_custom_field_by_name.return_value = outline_path
return mock_client
@patch("cheddahbot.tools.content_creation._get_clickup_client")
def test_phase2_detects_outline_approved_status(self, mock_get_client, tmp_db, tmp_path):
ctx, outline_path = self._setup_phase2(tmp_db, tmp_path)
mock_get_client.return_value = self._make_phase2_client(outline_path)
def test_phase2_detects_outline_review_state(self, tmp_db, tmp_path):
ctx = self._setup_phase2(tmp_db, tmp_path)
result = create_content(
url="https://example.com/plumbing",
keyword="plumbing services",
@ -333,8 +335,11 @@ class TestCreateContentPhase2:
)
assert "Phase 2 Complete" in result
def test_phase2_reads_outline(self, tmp_db, tmp_path):
ctx = self._setup_phase2(tmp_db, tmp_path)
@patch("cheddahbot.tools.content_creation._get_clickup_client")
def test_phase2_reads_outline(self, mock_get_client, tmp_db, tmp_path):
ctx, outline_path = self._setup_phase2(tmp_db, tmp_path)
mock_get_client.return_value = self._make_phase2_client(outline_path)
create_content(
url="https://example.com/plumbing",
keyword="plumbing services",
@ -344,8 +349,11 @@ class TestCreateContentPhase2:
prompt = call_args.args[0] if call_args.args else call_args.kwargs.get("prompt", "")
assert "Approved Outline" in prompt
def test_phase2_saves_content_file(self, tmp_db, tmp_path):
ctx = self._setup_phase2(tmp_db, tmp_path)
@patch("cheddahbot.tools.content_creation._get_clickup_client")
def test_phase2_saves_content_file(self, mock_get_client, tmp_db, tmp_path):
ctx, outline_path = self._setup_phase2(tmp_db, tmp_path)
mock_get_client.return_value = self._make_phase2_client(outline_path)
create_content(
url="https://example.com/plumbing",
keyword="plumbing services",
@ -355,20 +363,26 @@ class TestCreateContentPhase2:
assert content_file.exists()
assert content_file.read_text(encoding="utf-8") == "# Full Content\nParagraph..."
def test_phase2_sets_completed_state(self, tmp_db, tmp_path):
ctx = self._setup_phase2(tmp_db, tmp_path)
@patch("cheddahbot.tools.content_creation._get_clickup_client")
def test_phase2_syncs_clickup_complete(self, mock_get_client, tmp_db, tmp_path):
ctx, outline_path = self._setup_phase2(tmp_db, tmp_path)
mock_client = self._make_phase2_client(outline_path)
mock_get_client.return_value = mock_client
create_content(
url="https://example.com/plumbing",
keyword="plumbing services",
ctx=ctx,
)
raw = tmp_db.kv_get("clickup:task:task456:state")
state = json.loads(raw)
assert state["state"] == "completed"
assert "content_path" in state
# Verify ClickUp was synced to internal review
mock_client.update_task_status.assert_any_call("task456", "internal review")
mock_client.add_comment.assert_called()
@patch("cheddahbot.tools.content_creation._get_clickup_client")
def test_phase2_includes_clickup_sync_marker(self, mock_get_client, tmp_db, tmp_path):
ctx, outline_path = self._setup_phase2(tmp_db, tmp_path)
mock_get_client.return_value = self._make_phase2_client(outline_path)
def test_phase2_includes_clickup_sync_marker(self, tmp_db, tmp_path):
ctx = self._setup_phase2(tmp_db, tmp_path)
result = create_content(
url="https://example.com/plumbing",
keyword="plumbing services",
@ -392,9 +406,11 @@ class TestContinueContent:
result = continue_content(keyword="nonexistent", ctx=ctx)
assert "No outline awaiting review" in result
def test_finds_and_runs_phase2(self, tmp_db, tmp_path):
@patch("cheddahbot.tools.content_creation._get_clickup_client")
def test_finds_and_runs_phase2(self, mock_get_client, tmp_db, tmp_path):
cfg = Config()
cfg.content = ContentConfig(outline_dir=str(tmp_path / "outlines"))
cfg.clickup.space_id = "sp1"
# Create outline file
outline_dir = tmp_path / "outlines" / "plumbing-services"
@ -402,16 +418,17 @@ class TestContinueContent:
outline_file = outline_dir / "outline.md"
outline_file.write_text("## Outline", encoding="utf-8")
# Set kv state
state = {
"state": "outline_review",
"clickup_task_id": "task789",
"url": "https://example.com",
"keyword": "plumbing services",
"outline_path": str(outline_file),
"cora_path": "",
# Mock ClickUp client — returns a task matching the keyword
mock_client = MagicMock()
mock_task = MagicMock()
mock_task.id = "task789"
mock_task.custom_fields = {
"Keyword": "plumbing services",
"IMSURL": "https://example.com",
}
tmp_db.kv_set("clickup:task:task789:state", json.dumps(state))
mock_client.get_tasks_from_space.return_value = [mock_task]
mock_client.get_custom_field_by_name.return_value = str(outline_file)
mock_get_client.return_value = mock_client
agent = MagicMock()
agent.execute_task.return_value = "# Full content"
@ -426,7 +443,11 @@ class TestContinueContent:
class TestErrorPropagation:
def test_phase1_execution_error_sets_failed_state(self, tmp_db, tmp_path):
@patch("cheddahbot.tools.content_creation._get_clickup_client")
def test_phase1_execution_error_syncs_clickup(self, mock_get_client, tmp_db, tmp_path):
mock_client = MagicMock()
mock_get_client.return_value = mock_client
cfg = Config()
cfg.content = ContentConfig(outline_dir=str(tmp_path / "outlines"))
agent = MagicMock()
@ -443,11 +464,14 @@ class TestErrorPropagation:
ctx=ctx,
)
assert "Error:" in result
raw = tmp_db.kv_get("clickup:task:task_err:state")
state = json.loads(raw)
assert state["state"] == "failed"
# Verify ClickUp was notified of the failure
mock_client.update_task_status.assert_any_call("task_err", "error")
@patch("cheddahbot.tools.content_creation._get_clickup_client")
def test_phase1_error_return_syncs_clickup(self, mock_get_client, tmp_db, tmp_path):
mock_client = MagicMock()
mock_get_client.return_value = mock_client
def test_phase1_error_return_sets_failed(self, tmp_db, tmp_path):
cfg = Config()
cfg.content = ContentConfig(outline_dir=str(tmp_path / "outlines"))
agent = MagicMock()
@ -464,6 +488,5 @@ class TestErrorPropagation:
ctx=ctx,
)
assert result.startswith("Error:")
raw = tmp_db.kv_get("clickup:task:task_err2:state")
state = json.loads(raw)
assert state["state"] == "failed"
# Verify ClickUp was notified of the failure
mock_client.update_task_status.assert_any_call("task_err2", "error")

38
tests/test_linkbuilding.py
View File

@ -2,7 +2,6 @@
from __future__ import annotations
import json
import subprocess
from unittest.mock import MagicMock, patch
@ -548,16 +547,6 @@ class TestScanCoraFolder:
assert "Processed" in result
assert "old.xlsx" in result
def test_shows_kv_status(self, mock_ctx, tmp_path):
mock_ctx["config"].link_building.watch_folder = str(tmp_path)
(tmp_path / "tracked.xlsx").write_text("fake")
db = mock_ctx["db"]
db.kv_set("linkbuilding:watched:tracked.xlsx", json.dumps({"status": "completed"}))
result = scan_cora_folder(ctx=mock_ctx)
assert "completed" in result
# ---------------------------------------------------------------------------
# ClickUp state machine tests
@ -582,12 +571,6 @@ class TestClickUpStateMachine:
mock_ctx["clickup_task_id"] = "task_abc"
mock_ctx["config"].clickup.enabled = True
# Pre-set executing state
mock_ctx["db"].kv_set(
"clickup:task:task_abc:state",
json.dumps({"state": "executing"}),
)
ingest_proc = subprocess.CompletedProcess(
args=[], returncode=0, stdout=ingest_success_stdout, stderr=""
)
@ -603,10 +586,9 @@ class TestClickUpStateMachine:
assert "ClickUp Sync" in result
# Verify KV state was updated
raw = mock_ctx["db"].kv_get("clickup:task:task_abc:state")
state = json.loads(raw)
assert state["state"] == "completed"
# Verify ClickUp API was called for completion
cu.add_comment.assert_called()
cu.update_task_status.assert_called()
@patch("cheddahbot.tools.linkbuilding._run_blm_command")
@patch("cheddahbot.tools.linkbuilding._get_clickup_client")
@ -619,14 +601,6 @@ class TestClickUpStateMachine:
mock_ctx["clickup_task_id"] = "task_fail"
mock_ctx["config"].clickup.enabled = True
mock_ctx["config"].clickup.skill_map = {
"Link Building": {"error_status": "internal review"}
}
mock_ctx["db"].kv_set(
"clickup:task:task_fail:state",
json.dumps({"state": "executing"}),
)
mock_cmd.return_value = subprocess.CompletedProcess(
args=[], returncode=1, stdout="Error", stderr="crash"
@ -638,6 +612,6 @@ class TestClickUpStateMachine:
)
assert "Error" in result
raw = mock_ctx["db"].kv_get("clickup:task:task_fail:state")
state = json.loads(raw)
assert state["state"] == "failed"
# Verify ClickUp API was called for failure
cu.add_comment.assert_called()
cu.update_task_status.assert_called()

104
tests/test_scheduler.py
View File

@ -2,7 +2,6 @@
from __future__ import annotations
import json
from dataclasses import dataclass, field
from datetime import UTC, datetime
from unittest.mock import MagicMock
@ -104,55 +103,6 @@ class TestPollClickup:
mock_client.discover_field_filter.return_value = field_filter
return mock_client
def test_skips_task_already_completed(self, tmp_db):
"""Tasks with completed state should be skipped."""
config = _FakeConfig()
agent = MagicMock()
scheduler = Scheduler(config, tmp_db, agent)
state = {"state": "completed", "clickup_task_id": "t1"}
tmp_db.kv_set("clickup:task:t1:state", json.dumps(state))
due = str(_now_ms() + 86400000)
task = _make_task(
"t1",
"PR for Acme",
"Press Release",
due_date=due,
custom_fields=_FIELDS,
)
scheduler._clickup_client = self._make_mock_client(
tasks=[task],
)
scheduler._poll_clickup()
scheduler._clickup_client.update_task_status.assert_not_called()
def test_skips_task_already_failed(self, tmp_db):
"""Tasks with failed state should be skipped."""
config = _FakeConfig()
agent = MagicMock()
scheduler = Scheduler(config, tmp_db, agent)
state = {"state": "failed", "clickup_task_id": "t1"}
tmp_db.kv_set("clickup:task:t1:state", json.dumps(state))
due = str(_now_ms() + 86400000)
task = _make_task(
"t1",
"PR for Acme",
"Press Release",
due_date=due,
)
scheduler._clickup_client = self._make_mock_client(
tasks=[task],
)
scheduler._poll_clickup()
scheduler._clickup_client.update_task_status.assert_not_called()
def test_skips_task_with_no_due_date(self, tmp_db):
"""Tasks with no due date should be skipped."""
config = _FakeConfig()
@ -199,11 +149,11 @@ class TestExecuteTask:
"""Test the simplified _execute_task method."""
def test_success_flow(self, tmp_db):
"""Successful execution: state=completed."""
"""Successful execution: tool called, automation underway set."""
config = _FakeConfig()
agent = MagicMock()
agent._tools = MagicMock()
agent._tools.execute.return_value = "## ClickUp Sync\nDone"
agent._tools.execute.return_value = "Pipeline completed successfully"
scheduler = Scheduler(config, tmp_db, agent)
mock_client = MagicMock()
@ -224,51 +174,10 @@ class TestExecuteTask:
"t1",
"automation underway",
)
raw = tmp_db.kv_get("clickup:task:t1:state")
state = json.loads(raw)
assert state["state"] == "completed"
def test_success_fallback_path(self, tmp_db):
"""Scheduler uploads docx and sets review status."""
config = _FakeConfig()
agent = MagicMock()
agent._tools = MagicMock()
agent._tools.execute.return_value = "Press releases done.\n**Docx:** `output/pr.docx`"
scheduler = Scheduler(config, tmp_db, agent)
mock_client = MagicMock()
mock_client.update_task_status.return_value = True
mock_client.upload_attachment.return_value = True
mock_client.add_comment.return_value = True
scheduler._clickup_client = mock_client
due = str(_now_ms() + 86400000)
task = _make_task(
"t1",
"PR for Acme",
"Press Release",
due_date=due,
custom_fields=_FIELDS,
)
scheduler._execute_task(task)
mock_client.update_task_status.assert_any_call(
"t1",
"internal review",
)
mock_client.upload_attachment.assert_called_once_with(
"t1",
"output/pr.docx",
)
raw = tmp_db.kv_get("clickup:task:t1:state")
state = json.loads(raw)
assert state["state"] == "completed"
assert "output/pr.docx" in state["deliverable_paths"]
agent._tools.execute.assert_called_once()
def test_failure_flow(self, tmp_db):
"""Failed: state=failed, error comment, status set to 'error'."""
"""Failed: error comment posted, status set to 'error'."""
config = _FakeConfig()
agent = MagicMock()
agent._tools = MagicMock()
@ -295,11 +204,6 @@ class TestExecuteTask:
comment_text = mock_client.add_comment.call_args[0][1]
assert "failed" in comment_text.lower()
raw = tmp_db.kv_get("clickup:task:t1:state")
state = json.loads(raw)
assert state["state"] == "failed"
assert "API timeout" in state["error"]
class TestFieldFilterDiscovery:
"""Test _discover_field_filter caching."""

54
tests/test_scheduler_helpers.py
View File

@ -1,32 +1,42 @@
"""Tests for scheduler helper functions."""
"""Tests for scheduler helper functions.
Note: _extract_docx_paths was removed as part of KV store elimination.
The scheduler no longer handles docx extraction — tools own their own sync.
"""
from __future__ import annotations
from cheddahbot.scheduler import _extract_docx_paths
class TestLoopTimestamps:
"""Test that loop timestamps use in-memory storage."""
class TestExtractDocxPaths:
def test_extracts_paths_from_realistic_output(self):
result = (
"Press releases generated successfully!\n\n"
"**Docx:** `output/press_releases/acme-corp-launch.docx`\n"
"**Docx:** `output/press_releases/acme-corp-expansion.docx`\n"
"Files saved to output/press_releases/"
)
paths = _extract_docx_paths(result)
def test_initial_timestamps_are_none(self):
from unittest.mock import MagicMock
assert len(paths) == 2
assert paths[0] == "output/press_releases/acme-corp-launch.docx"
assert paths[1] == "output/press_releases/acme-corp-expansion.docx"
from cheddahbot.scheduler import Scheduler
def test_returns_empty_list_when_no_paths(self):
result = "Task completed successfully. No files generated."
paths = _extract_docx_paths(result)
config = MagicMock()
db = MagicMock()
agent = MagicMock()
sched = Scheduler(config, db, agent)
assert paths == []
timestamps = sched.get_loop_timestamps()
assert timestamps["heartbeat"] is None
assert timestamps["poll"] is None
assert timestamps["clickup"] is None
def test_only_matches_docx_extension(self):
result = "**Docx:** `report.docx`\n**PDF:** `report.pdf`\n**Docx:** `summary.txt`\n"
paths = _extract_docx_paths(result)
def test_timestamps_update_in_memory(self):
from unittest.mock import MagicMock
assert paths == ["report.docx"]
from cheddahbot.scheduler import Scheduler
config = MagicMock()
db = MagicMock()
agent = MagicMock()
sched = Scheduler(config, db, agent)
sched._loop_timestamps["heartbeat"] = "2026-02-27T12:00:00+00:00"
timestamps = sched.get_loop_timestamps()
assert timestamps["heartbeat"] == "2026-02-27T12:00:00+00:00"
# Ensure db.kv_set was never called
db.kv_set.assert_not_called()