Add two-phase content creation tool with human-in-the-loop outline review
Phase 1 researches competitors and generates an outline via the execution brain, saves it to a network/local path, and pauses for human review. Phase 2 picks up the approved outline and writes full SEO-optimized content. ClickUp integration maps "On Page Optimization" and "Content Creation" work categories, with "outline approved" added to poll_statuses for automatic Phase 2 triggering. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>content-creation
parent
0e3e3bc945
commit
a3b8457afe
|
|
@ -369,6 +369,7 @@ class Agent:
|
|||
system_context: str = "",
|
||||
tools: str = "",
|
||||
model: str = "",
|
||||
skip_permissions: bool = False,
|
||||
) -> str:
|
||||
"""Execute a task using the execution brain (Claude Code CLI).
|
||||
|
||||
|
|
@ -378,6 +379,7 @@ class Agent:
|
|||
Args:
|
||||
tools: Override Claude Code tool list (e.g. "Bash,Read,WebSearch").
|
||||
model: Override the CLI model (e.g. "claude-sonnet-4.5").
|
||||
skip_permissions: If True, run CLI with --dangerously-skip-permissions.
|
||||
"""
|
||||
log.info("Execution brain task: %s", prompt[:100])
|
||||
kwargs: dict = {"system_prompt": system_context}
|
||||
|
|
@ -385,6 +387,8 @@ class Agent:
|
|||
kwargs["tools"] = tools
|
||||
if model:
|
||||
kwargs["model"] = model
|
||||
if skip_permissions:
|
||||
kwargs["skip_permissions"] = True
|
||||
result = self.llm.execute(prompt, **kwargs)
|
||||
|
||||
# Log to daily memory
|
||||
|
|
|
|||
|
|
@ -82,6 +82,15 @@ class ApiBudgetConfig:
|
|||
alert_threshold: float = 0.8 # alert at 80% of limit
|
||||
|
||||
|
||||
@dataclass
|
||||
class ContentConfig:
|
||||
cora_inbox: str = "" # e.g. "Z:/content-cora-inbox"
|
||||
outline_dir: str = "" # e.g. "Z:/content-outlines"
|
||||
company_capabilities_default: str = (
|
||||
"All certifications and licenses need to be verified on the company's website."
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class AgentConfig:
|
||||
"""Per-agent configuration for multi-agent support."""
|
||||
|
|
@ -112,6 +121,7 @@ class Config:
|
|||
email: EmailConfig = field(default_factory=EmailConfig)
|
||||
link_building: LinkBuildingConfig = field(default_factory=LinkBuildingConfig)
|
||||
api_budget: ApiBudgetConfig = field(default_factory=ApiBudgetConfig)
|
||||
content: ContentConfig = field(default_factory=ContentConfig)
|
||||
agents: list[AgentConfig] = field(default_factory=lambda: [AgentConfig()])
|
||||
|
||||
# Derived paths
|
||||
|
|
@ -167,6 +177,10 @@ def load_config() -> Config:
|
|||
for k, v in data["api_budget"].items():
|
||||
if hasattr(cfg.api_budget, k):
|
||||
setattr(cfg.api_budget, k, v)
|
||||
if "content" in data and isinstance(data["content"], dict):
|
||||
for k, v in data["content"].items():
|
||||
if hasattr(cfg.content, k):
|
||||
setattr(cfg.content, k, v)
|
||||
|
||||
# Multi-agent configs
|
||||
if "agents" in data and isinstance(data["agents"], list):
|
||||
|
|
|
|||
|
|
@ -156,6 +156,7 @@ class LLMAdapter:
|
|||
working_dir: str | None = None,
|
||||
tools: str = "Bash,Read,Edit,Write,Glob,Grep",
|
||||
model: str | None = None,
|
||||
skip_permissions: bool = False,
|
||||
) -> str:
|
||||
"""Execution brain: calls Claude Code CLI with full tool access.
|
||||
|
||||
|
|
@ -165,6 +166,8 @@ class LLMAdapter:
|
|||
Args:
|
||||
tools: Comma-separated Claude Code tool names (default: standard set).
|
||||
model: Override the CLI model (e.g. "claude-sonnet-4.5").
|
||||
skip_permissions: If True, append --dangerously-skip-permissions to
|
||||
the CLI invocation (used for automated pipelines).
|
||||
"""
|
||||
claude_bin = shutil.which("claude")
|
||||
if not claude_bin:
|
||||
|
|
@ -188,6 +191,8 @@ class LLMAdapter:
|
|||
cmd.extend(["--model", model])
|
||||
if system_prompt:
|
||||
cmd.extend(["--system-prompt", system_prompt])
|
||||
if skip_permissions:
|
||||
cmd.append("--dangerously-skip-permissions")
|
||||
|
||||
log.debug("Execution brain cmd: %s", " ".join(cmd[:6]) + "...")
|
||||
|
||||
|
|
@ -355,9 +360,14 @@ class LLMAdapter:
|
|||
|
||||
except Exception as e:
|
||||
if not has_yielded and attempt < max_retries and _is_retryable_error(e):
|
||||
wait = 2 ** attempt
|
||||
log.warning("Retryable LLM error (attempt %d/%d), retrying in %ds: %s",
|
||||
attempt + 1, max_retries + 1, wait, e)
|
||||
wait = 2**attempt
|
||||
log.warning(
|
||||
"Retryable LLM error (attempt %d/%d), retrying in %ds: %s",
|
||||
attempt + 1,
|
||||
max_retries + 1,
|
||||
wait,
|
||||
e,
|
||||
)
|
||||
time.sleep(wait)
|
||||
continue
|
||||
yield {"type": "text", "content": _friendly_error(e, self.provider)}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,620 @@
|
|||
"""Two-phase content creation pipeline tool.
|
||||
|
||||
Phase 1: Research competitors + generate outline → save → stop for human review.
|
||||
Phase 2: Human approves/edits outline → tool picks it up → writes full content.
|
||||
|
||||
The content-researcher skill in the execution brain is triggered by keywords like
|
||||
"service page", "content optimization", "SEO content", etc.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
from datetime import UTC, datetime
|
||||
from pathlib import Path
|
||||
|
||||
from . import tool
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
_ROOT_DIR = Path(__file__).resolve().parent.parent.parent
|
||||
_DATA_DIR = _ROOT_DIR / "data"
|
||||
_LOCAL_CONTENT_DIR = _DATA_DIR / "generated" / "content"
|
||||
|
||||
EXEC_TOOLS = "Bash,Read,Edit,Write,Glob,Grep,WebSearch,WebFetch"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ClickUp helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _get_clickup_client(ctx: dict | None):
|
||||
"""Create a ClickUpClient from tool context, or None if unavailable."""
|
||||
if not ctx or not ctx.get("config") or not ctx["config"].clickup.enabled:
|
||||
return None
|
||||
try:
|
||||
from ..clickup import ClickUpClient
|
||||
|
||||
config = ctx["config"]
|
||||
return ClickUpClient(
|
||||
api_token=config.clickup.api_token,
|
||||
workspace_id=config.clickup.workspace_id,
|
||||
task_type_field_name=config.clickup.task_type_field_name,
|
||||
)
|
||||
except Exception as e:
|
||||
log.warning("Could not create ClickUp client: %s", e)
|
||||
return None
|
||||
|
||||
|
||||
def _sync_clickup_start(ctx: dict | None, task_id: str) -> None:
|
||||
"""Move ClickUp task to 'automation underway'."""
|
||||
if not task_id or not ctx:
|
||||
return
|
||||
client = _get_clickup_client(ctx)
|
||||
if not client:
|
||||
return
|
||||
try:
|
||||
config = ctx["config"]
|
||||
client.update_task_status(task_id, config.clickup.automation_status)
|
||||
except Exception as e:
|
||||
log.warning("Failed to set ClickUp start status for %s: %s", task_id, e)
|
||||
finally:
|
||||
client.close()
|
||||
|
||||
|
||||
def _sync_clickup_outline_ready(ctx: dict | None, task_id: str, outline_path: str) -> None:
|
||||
"""Post outline comment and move ClickUp task to 'outline review'."""
|
||||
if not task_id or not ctx:
|
||||
return
|
||||
client = _get_clickup_client(ctx)
|
||||
if not client:
|
||||
return
|
||||
try:
|
||||
client.add_comment(
|
||||
task_id,
|
||||
f"📝 CheddahBot generated a content outline.\n\n"
|
||||
f"Outline saved to: `{outline_path}`\n\n"
|
||||
f"Please review and edit the outline, then move this task to "
|
||||
f"**outline approved** to trigger the full content write.",
|
||||
)
|
||||
client.update_task_status(task_id, "outline review")
|
||||
except Exception as e:
|
||||
log.warning("Failed to sync outline-ready for %s: %s", task_id, e)
|
||||
finally:
|
||||
client.close()
|
||||
|
||||
|
||||
def _sync_clickup_complete(ctx: dict | None, task_id: str, content_path: str) -> None:
|
||||
"""Post completion comment and move ClickUp task to 'internal review'."""
|
||||
if not task_id or not ctx:
|
||||
return
|
||||
client = _get_clickup_client(ctx)
|
||||
if not client:
|
||||
return
|
||||
try:
|
||||
config = ctx["config"]
|
||||
client.add_comment(
|
||||
task_id,
|
||||
f"✅ CheddahBot completed the content.\n\n"
|
||||
f"Final content saved to: `{content_path}`\n\n"
|
||||
f"Ready for internal review.",
|
||||
)
|
||||
client.update_task_status(task_id, config.clickup.review_status)
|
||||
except Exception as e:
|
||||
log.warning("Failed to sync completion for %s: %s", task_id, e)
|
||||
finally:
|
||||
client.close()
|
||||
|
||||
|
||||
def _sync_clickup_fail(ctx: dict | None, task_id: str, error: str) -> None:
|
||||
"""Post error comment and move ClickUp task to 'error'."""
|
||||
if not task_id or not ctx:
|
||||
return
|
||||
client = _get_clickup_client(ctx)
|
||||
if not client:
|
||||
return
|
||||
try:
|
||||
config = ctx["config"]
|
||||
client.add_comment(
|
||||
task_id,
|
||||
f"❌ CheddahBot failed during content creation.\n\nError: {error[:2000]}",
|
||||
)
|
||||
client.update_task_status(task_id, config.clickup.error_status)
|
||||
except Exception as e:
|
||||
log.warning("Failed to sync failure for %s: %s", task_id, e)
|
||||
finally:
|
||||
client.close()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _slugify(text: str) -> str:
|
||||
"""Turn text into a filesystem-safe slug."""
|
||||
text = text.lower().strip()
|
||||
text = re.sub(r"[^\w\s-]", "", text)
|
||||
text = re.sub(r"[\s_]+", "-", text)
|
||||
return text[:80].strip("-")
|
||||
|
||||
|
||||
def _find_cora_report(keyword: str, cora_inbox: str) -> str:
|
||||
"""Fuzzy-match a Cora .xlsx report by keyword.
|
||||
|
||||
Match priority: exact filename match > substring > word overlap.
|
||||
Skips Office temp files (~$...).
|
||||
Returns the path string, or "" if not found.
|
||||
"""
|
||||
if not cora_inbox or not keyword:
|
||||
return ""
|
||||
inbox = Path(cora_inbox)
|
||||
if not inbox.exists():
|
||||
return ""
|
||||
|
||||
xlsx_files = [f for f in inbox.glob("*.xlsx") if not f.name.startswith("~$")]
|
||||
if not xlsx_files:
|
||||
return ""
|
||||
|
||||
keyword_lower = keyword.lower().strip()
|
||||
keyword_words = set(keyword_lower.split())
|
||||
|
||||
# Pass 1: exact stem match
|
||||
for f in xlsx_files:
|
||||
if f.stem.lower().strip() == keyword_lower:
|
||||
return str(f)
|
||||
|
||||
# Pass 2: keyword is substring of filename (or vice versa)
|
||||
for f in xlsx_files:
|
||||
stem = f.stem.lower().strip()
|
||||
if keyword_lower in stem or stem in keyword_lower:
|
||||
return str(f)
|
||||
|
||||
# Pass 3: word overlap (at least half the keyword words)
|
||||
best_match = ""
|
||||
best_overlap = 0
|
||||
for f in xlsx_files:
|
||||
stem_words = set(f.stem.lower().replace("-", " ").replace("_", " ").split())
|
||||
overlap = len(keyword_words & stem_words)
|
||||
if overlap > best_overlap and overlap >= max(1, len(keyword_words) // 2):
|
||||
best_overlap = overlap
|
||||
best_match = str(f)
|
||||
|
||||
return best_match
|
||||
|
||||
|
||||
def _save_content(content: str, keyword: str, filename: str, config) -> str:
|
||||
"""Save content to the outline directory (network path with local fallback).
|
||||
|
||||
Returns the actual path used.
|
||||
"""
|
||||
slug = _slugify(keyword)
|
||||
if not slug:
|
||||
slug = "unknown"
|
||||
|
||||
# Try primary (network) path
|
||||
if config.content.outline_dir:
|
||||
primary = Path(config.content.outline_dir) / slug
|
||||
try:
|
||||
primary.mkdir(parents=True, exist_ok=True)
|
||||
out_path = primary / filename
|
||||
out_path.write_text(content, encoding="utf-8")
|
||||
return str(out_path)
|
||||
except OSError as e:
|
||||
log.warning("Network path unavailable (%s), falling back to local: %s", primary, e)
|
||||
|
||||
# Fallback to local
|
||||
local = _LOCAL_CONTENT_DIR / slug
|
||||
local.mkdir(parents=True, exist_ok=True)
|
||||
out_path = local / filename
|
||||
out_path.write_text(content, encoding="utf-8")
|
||||
return str(out_path)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Prompt builders
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _build_phase1_prompt(
|
||||
url: str,
|
||||
keyword: str,
|
||||
content_type: str,
|
||||
cora_path: str,
|
||||
capabilities_default: str,
|
||||
) -> str:
|
||||
"""Build the Phase 1 prompt that triggers the content-researcher skill."""
|
||||
parts = [
|
||||
f"Research, outline, and draft an optimized {content_type} for {url} "
|
||||
f"targeting keyword '{keyword}'. This is an SEO content optimization project.",
|
||||
]
|
||||
if cora_path:
|
||||
parts.append(
|
||||
f"\nA Cora SEO report is available at: {cora_path}\n"
|
||||
f"Read this report to extract keyword targets, entity requirements, "
|
||||
f"and competitive analysis data."
|
||||
)
|
||||
if capabilities_default:
|
||||
parts.append(
|
||||
f'\nWhen asked about company capabilities, respond with: "{capabilities_default}"'
|
||||
)
|
||||
parts.append(
|
||||
"\nDeliver the outline as a complete markdown document with sections, "
|
||||
"headings, entity targets, and keyword placement notes."
|
||||
)
|
||||
return "\n".join(parts)
|
||||
|
||||
|
||||
def _build_phase2_prompt(
|
||||
url: str,
|
||||
keyword: str,
|
||||
outline_text: str,
|
||||
cora_path: str,
|
||||
) -> str:
|
||||
"""Build the Phase 2 prompt for writing full content from an approved outline."""
|
||||
parts = [
|
||||
f"Write full SEO-optimized content based on this approved outline for {url} "
|
||||
f"targeting '{keyword}'. This is the content writing phase of a "
|
||||
f"content optimization project.",
|
||||
f"\n## Approved Outline\n\n{outline_text}",
|
||||
]
|
||||
if cora_path:
|
||||
parts.append(
|
||||
f"\nThe Cora SEO report is at: {cora_path}\n"
|
||||
f"Use it for keyword density targets and entity optimization."
|
||||
)
|
||||
parts.append(
|
||||
"\nWrite publication-ready content following the outline structure. "
|
||||
"Include all entity targets and keyword placements as noted in the outline."
|
||||
)
|
||||
return "\n".join(parts)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Main tool
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@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.",
|
||||
category="content",
|
||||
)
|
||||
def create_content(
|
||||
url: str,
|
||||
keyword: str,
|
||||
content_type: str = "service page",
|
||||
ctx: dict | None = None,
|
||||
) -> str:
|
||||
"""Create SEO content in two phases with human review between them.
|
||||
|
||||
Args:
|
||||
url: Target page URL (e.g. "https://example.com/services/plumbing").
|
||||
keyword: Primary target keyword (e.g. "plumbing services").
|
||||
content_type: Type of content — "service page", "blog post", etc.
|
||||
"""
|
||||
if not url or not keyword:
|
||||
return "Error: Both 'url' and 'keyword' are required."
|
||||
if not ctx or "agent" not in ctx:
|
||||
return "Error: Tool context with agent is required."
|
||||
|
||||
agent = ctx["agent"]
|
||||
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
|
||||
phase = 1
|
||||
existing_state = {}
|
||||
if kv_key and db:
|
||||
raw = db.kv_get(kv_key)
|
||||
if raw:
|
||||
try:
|
||||
existing_state = json.loads(raw)
|
||||
if existing_state.get("state") == "outline_review":
|
||||
phase = 2
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
|
||||
# Find Cora report
|
||||
cora_inbox = config.content.cora_inbox if config else ""
|
||||
cora_path = _find_cora_report(keyword, cora_inbox)
|
||||
if cora_path:
|
||||
log.info("Found Cora report for '%s': %s", keyword, cora_path)
|
||||
|
||||
capabilities_default = config.content.company_capabilities_default if config else ""
|
||||
|
||||
if phase == 1:
|
||||
return _run_phase1(
|
||||
agent=agent,
|
||||
config=config,
|
||||
db=db,
|
||||
ctx=ctx,
|
||||
task_id=task_id,
|
||||
kv_key=kv_key,
|
||||
url=url,
|
||||
keyword=keyword,
|
||||
content_type=content_type,
|
||||
cora_path=cora_path,
|
||||
capabilities_default=capabilities_default,
|
||||
)
|
||||
else:
|
||||
return _run_phase2(
|
||||
agent=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=existing_state,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Phase 1: Research + Outline
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _run_phase1(
|
||||
*,
|
||||
agent,
|
||||
config,
|
||||
db,
|
||||
ctx,
|
||||
task_id: str,
|
||||
kv_key: str,
|
||||
url: str,
|
||||
keyword: str,
|
||||
content_type: str,
|
||||
cora_path: str,
|
||||
capabilities_default: str,
|
||||
) -> str:
|
||||
now = datetime.now(UTC).isoformat()
|
||||
|
||||
# ClickUp: move to automation underway
|
||||
if task_id:
|
||||
_sync_clickup_start(ctx, task_id)
|
||||
|
||||
prompt = _build_phase1_prompt(url, keyword, content_type, cora_path, capabilities_default)
|
||||
|
||||
log.info("Phase 1 — researching + outlining for '%s' (%s)", keyword, url)
|
||||
try:
|
||||
result = agent.execute_task(
|
||||
prompt,
|
||||
tools=EXEC_TOOLS,
|
||||
skip_permissions=True,
|
||||
)
|
||||
except Exception as e:
|
||||
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
|
||||
|
||||
# Save the outline
|
||||
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
|
||||
if task_id:
|
||||
_sync_clickup_outline_ready(ctx, task_id, outline_path)
|
||||
|
||||
return (
|
||||
f"## Phase 1 Complete — Outline Ready for Review\n\n"
|
||||
f"**Keyword:** {keyword}\n"
|
||||
f"**URL:** {url}\n"
|
||||
f"**Outline saved to:** `{outline_path}`\n\n"
|
||||
f"Please review and edit the outline. When ready, move the ClickUp task "
|
||||
f"to **outline approved** to trigger Phase 2 (full content writing).\n\n"
|
||||
f"---\n\n{result}\n\n"
|
||||
f"## ClickUp Sync\nPhase 1 complete. Status: outline review."
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Phase 2: Write Full Content
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _run_phase2(
|
||||
*,
|
||||
agent,
|
||||
config,
|
||||
db,
|
||||
ctx,
|
||||
task_id: str,
|
||||
kv_key: str,
|
||||
url: str,
|
||||
keyword: str,
|
||||
cora_path: str,
|
||||
existing_state: dict,
|
||||
) -> str:
|
||||
# Read the (possibly edited) outline
|
||||
outline_path = existing_state.get("outline_path", "")
|
||||
outline_text = ""
|
||||
if outline_path:
|
||||
try:
|
||||
outline_text = Path(outline_path).read_text(encoding="utf-8")
|
||||
except OSError as e:
|
||||
log.warning("Could not read outline at %s: %s", outline_path, e)
|
||||
|
||||
if not outline_text:
|
||||
return (
|
||||
"Error: Could not read the outline file. "
|
||||
f"Expected at: {outline_path or '(no path saved)'}"
|
||||
)
|
||||
|
||||
# Use saved cora_path from state if we don't have one now
|
||||
if not cora_path:
|
||||
cora_path = existing_state.get("cora_path", "")
|
||||
|
||||
# ClickUp: move to automation underway
|
||||
if task_id:
|
||||
_sync_clickup_start(ctx, task_id)
|
||||
|
||||
prompt = _build_phase2_prompt(url, keyword, outline_text, cora_path)
|
||||
|
||||
log.info("Phase 2 — writing full content for '%s' (%s)", keyword, url)
|
||||
try:
|
||||
result = agent.execute_task(
|
||||
prompt,
|
||||
tools=EXEC_TOOLS,
|
||||
skip_permissions=True,
|
||||
)
|
||||
except Exception as e:
|
||||
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
|
||||
|
||||
# Save final content
|
||||
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)
|
||||
|
||||
return (
|
||||
f"## Phase 2 Complete — Content Written\n\n"
|
||||
f"**Keyword:** {keyword}\n"
|
||||
f"**URL:** {url}\n"
|
||||
f"**Content saved to:** `{content_path}`\n\n"
|
||||
f"---\n\n{result}\n\n"
|
||||
f"## ClickUp Sync\nPhase 2 complete. Status: internal review."
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Continue content (chat-initiated Phase 2)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@tool(
|
||||
"continue_content",
|
||||
"Resume content creation for a keyword that has an approved outline. "
|
||||
"Runs Phase 2 (full content writing) for a previously outlined keyword.",
|
||||
category="content",
|
||||
)
|
||||
def continue_content(
|
||||
keyword: str,
|
||||
ctx: dict | None = None,
|
||||
) -> str:
|
||||
"""Resume content writing for a keyword with an approved outline.
|
||||
|
||||
Args:
|
||||
keyword: The keyword to continue writing content for.
|
||||
"""
|
||||
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."
|
||||
|
||||
db = ctx["db"]
|
||||
config = ctx.get("config")
|
||||
|
||||
# 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:
|
||||
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", "")
|
||||
|
||||
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 (
|
||||
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))
|
||||
24
config.yaml
24
config.yaml
|
|
@ -42,7 +42,7 @@ email:
|
|||
# ClickUp integration
|
||||
clickup:
|
||||
poll_interval_minutes: 20 # 3x per hour
|
||||
poll_statuses: ["to do"]
|
||||
poll_statuses: ["to do", "outline approved"]
|
||||
review_status: "internal review"
|
||||
in_progress_status: "in progress"
|
||||
automation_status: "automation underway"
|
||||
|
|
@ -58,6 +58,18 @@ clickup:
|
|||
company_name: "Customer"
|
||||
target_url: "IMSURL"
|
||||
branded_url: "SocialURL"
|
||||
"On Page Optimization":
|
||||
tool: "create_content"
|
||||
auto_execute: true
|
||||
field_mapping:
|
||||
url: "IMSURL"
|
||||
keyword: "Keyword"
|
||||
"Content Creation":
|
||||
tool: "create_content"
|
||||
auto_execute: true
|
||||
field_mapping:
|
||||
url: "IMSURL"
|
||||
keyword: "Keyword"
|
||||
"Link Building":
|
||||
tool: "run_link_building"
|
||||
auto_execute: false
|
||||
|
|
@ -79,6 +91,11 @@ link_building:
|
|||
watch_interval_minutes: 60
|
||||
default_branded_plus_ratio: 0.7
|
||||
|
||||
# Content creation settings
|
||||
content:
|
||||
cora_inbox: "Z:/content-cora-inbox"
|
||||
outline_dir: "Z:/content-outlines"
|
||||
|
||||
# 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.
|
||||
|
|
@ -113,6 +130,11 @@ agents:
|
|||
tools: [run_link_building, run_cora_backlinks, blm_ingest_cora, blm_generate_batch, scan_cora_folder, delegate_task, remember, search_memory]
|
||||
memory_scope: ""
|
||||
|
||||
- name: content_creator
|
||||
display_name: Content Creator
|
||||
tools: [create_content, continue_content, delegate_task, remember, search_memory, web_search, web_fetch]
|
||||
memory_scope: ""
|
||||
|
||||
- name: planner
|
||||
display_name: Planner
|
||||
model: "x-ai/grok-4.1-fast"
|
||||
|
|
|
|||
|
|
@ -0,0 +1,471 @@
|
|||
"""Tests for the content creation pipeline tool."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from cheddahbot.config import Config, ContentConfig
|
||||
from cheddahbot.tools.content_creation import (
|
||||
_build_phase1_prompt,
|
||||
_build_phase2_prompt,
|
||||
_find_cora_report,
|
||||
_save_content,
|
||||
_slugify,
|
||||
continue_content,
|
||||
create_content,
|
||||
)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _slugify
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_slugify_basic():
|
||||
assert _slugify("Plumbing Services") == "plumbing-services"
|
||||
|
||||
|
||||
def test_slugify_special_chars():
|
||||
assert _slugify("AC Repair & Maintenance!") == "ac-repair-maintenance"
|
||||
|
||||
|
||||
def test_slugify_truncates():
|
||||
long = "a" * 200
|
||||
assert len(_slugify(long)) <= 80
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _build_phase1_prompt
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestBuildPhase1Prompt:
|
||||
def test_contains_trigger_keywords(self):
|
||||
prompt = _build_phase1_prompt(
|
||||
"https://example.com/plumbing",
|
||||
"plumbing services",
|
||||
"service page",
|
||||
"",
|
||||
"",
|
||||
)
|
||||
assert "SEO content optimization" in prompt
|
||||
assert "plumbing services" in prompt
|
||||
assert "service page" in prompt
|
||||
assert "https://example.com/plumbing" in prompt
|
||||
|
||||
def test_includes_cora_path(self):
|
||||
prompt = _build_phase1_prompt(
|
||||
"https://example.com",
|
||||
"keyword",
|
||||
"blog post",
|
||||
"Z:/cora/report.xlsx",
|
||||
"",
|
||||
)
|
||||
assert "Z:/cora/report.xlsx" in prompt
|
||||
assert "Cora SEO report" in prompt
|
||||
|
||||
def test_includes_capabilities_default(self):
|
||||
default = "Verify on website."
|
||||
prompt = _build_phase1_prompt(
|
||||
"https://example.com",
|
||||
"keyword",
|
||||
"service page",
|
||||
"",
|
||||
default,
|
||||
)
|
||||
assert default in prompt
|
||||
assert "company capabilities" in prompt
|
||||
|
||||
def test_no_cora_no_capabilities(self):
|
||||
prompt = _build_phase1_prompt(
|
||||
"https://example.com",
|
||||
"keyword",
|
||||
"service page",
|
||||
"",
|
||||
"",
|
||||
)
|
||||
assert "Cora SEO report" not in prompt
|
||||
assert "company capabilities" not in prompt
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _build_phase2_prompt
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestBuildPhase2Prompt:
|
||||
def test_contains_outline(self):
|
||||
outline = "## Section 1\nContent here."
|
||||
prompt = _build_phase2_prompt(
|
||||
"https://example.com",
|
||||
"plumbing",
|
||||
outline,
|
||||
"",
|
||||
)
|
||||
assert outline in prompt
|
||||
assert "content writing phase" in prompt
|
||||
assert "plumbing" in prompt
|
||||
|
||||
def test_includes_cora_path(self):
|
||||
prompt = _build_phase2_prompt(
|
||||
"https://example.com",
|
||||
"keyword",
|
||||
"outline text",
|
||||
"Z:/cora/report.xlsx",
|
||||
)
|
||||
assert "Z:/cora/report.xlsx" in prompt
|
||||
|
||||
def test_no_cora(self):
|
||||
prompt = _build_phase2_prompt(
|
||||
"https://example.com",
|
||||
"keyword",
|
||||
"outline text",
|
||||
"",
|
||||
)
|
||||
assert "Cora SEO report" not in prompt
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _find_cora_report
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestFindCoraReport:
|
||||
def test_empty_inbox(self, tmp_path):
|
||||
assert _find_cora_report("keyword", str(tmp_path)) == ""
|
||||
|
||||
def test_nonexistent_path(self):
|
||||
assert _find_cora_report("keyword", "/nonexistent/path") == ""
|
||||
|
||||
def test_empty_keyword(self, tmp_path):
|
||||
assert _find_cora_report("", str(tmp_path)) == ""
|
||||
|
||||
def test_exact_match(self, tmp_path):
|
||||
report = tmp_path / "plumbing services.xlsx"
|
||||
report.touch()
|
||||
result = _find_cora_report("plumbing services", str(tmp_path))
|
||||
assert result == str(report)
|
||||
|
||||
def test_substring_match(self, tmp_path):
|
||||
report = tmp_path / "plumbing-services-city.xlsx"
|
||||
report.touch()
|
||||
result = _find_cora_report("plumbing services", str(tmp_path))
|
||||
# "plumbing services" is a substring of "plumbing-services-city"
|
||||
assert result == str(report)
|
||||
|
||||
def test_word_overlap(self, tmp_path):
|
||||
report = tmp_path / "residential-plumbing-repair.xlsx"
|
||||
report.touch()
|
||||
result = _find_cora_report("plumbing repair", str(tmp_path))
|
||||
assert result == str(report)
|
||||
|
||||
def test_skips_temp_files(self, tmp_path):
|
||||
(tmp_path / "~$report.xlsx").touch()
|
||||
(tmp_path / "actual-report.xlsx").touch()
|
||||
result = _find_cora_report("actual report", str(tmp_path))
|
||||
assert "~$" not in result
|
||||
assert "actual-report" in result
|
||||
|
||||
def test_no_match(self, tmp_path):
|
||||
(tmp_path / "completely-unrelated.xlsx").touch()
|
||||
result = _find_cora_report("plumbing services", str(tmp_path))
|
||||
assert result == ""
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _save_content
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestSaveContent:
|
||||
def _make_config(self, outline_dir: str = "") -> Config:
|
||||
cfg = Config()
|
||||
cfg.content = ContentConfig(outline_dir=outline_dir)
|
||||
return cfg
|
||||
|
||||
def test_saves_to_primary_path(self, tmp_path):
|
||||
cfg = self._make_config(str(tmp_path / "outlines"))
|
||||
path = _save_content("# Outline", "plumbing services", "outline.md", cfg)
|
||||
assert "outlines" in path
|
||||
assert Path(path).read_text(encoding="utf-8") == "# Outline"
|
||||
|
||||
def test_falls_back_to_local(self, tmp_path):
|
||||
# Point to an invalid network path
|
||||
cfg = self._make_config("\\\\nonexistent\\share\\outlines")
|
||||
with patch(
|
||||
"cheddahbot.tools.content_creation._LOCAL_CONTENT_DIR",
|
||||
tmp_path / "local",
|
||||
):
|
||||
path = _save_content("# Outline", "plumbing", "outline.md", cfg)
|
||||
assert str(tmp_path / "local") in path
|
||||
assert Path(path).read_text(encoding="utf-8") == "# Outline"
|
||||
|
||||
def test_empty_outline_dir_uses_local(self, tmp_path):
|
||||
cfg = self._make_config("")
|
||||
with patch(
|
||||
"cheddahbot.tools.content_creation._LOCAL_CONTENT_DIR",
|
||||
tmp_path / "local",
|
||||
):
|
||||
path = _save_content("content", "keyword", "outline.md", cfg)
|
||||
assert str(tmp_path / "local") in path
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# create_content — Phase 1
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestCreateContentPhase1:
|
||||
def _make_ctx(self, tmp_db, tmp_path):
|
||||
cfg = Config()
|
||||
cfg.content = ContentConfig(outline_dir=str(tmp_path / "outlines"))
|
||||
agent = MagicMock()
|
||||
agent.execute_task.return_value = "## Generated Outline\nSection 1..."
|
||||
return {
|
||||
"agent": agent,
|
||||
"config": cfg,
|
||||
"db": tmp_db,
|
||||
"clickup_task_id": "task123",
|
||||
}
|
||||
|
||||
def test_requires_url_and_keyword(self, tmp_db):
|
||||
ctx = {"agent": MagicMock(), "config": Config(), "db": tmp_db}
|
||||
assert create_content(url="", keyword="test", ctx=ctx).startswith("Error:")
|
||||
assert create_content(url="http://x", keyword="", ctx=ctx).startswith("Error:")
|
||||
|
||||
def test_requires_context(self):
|
||||
assert create_content(url="http://x", keyword="kw", ctx=None).startswith("Error:")
|
||||
|
||||
def test_phase1_runs_without_prior_state(self, tmp_db, tmp_path):
|
||||
ctx = self._make_ctx(tmp_db, tmp_path)
|
||||
result = create_content(
|
||||
url="https://example.com/services",
|
||||
keyword="plumbing services",
|
||||
ctx=ctx,
|
||||
)
|
||||
assert "Phase 1 Complete" in result
|
||||
assert "outline" in result.lower()
|
||||
ctx["agent"].execute_task.assert_called_once()
|
||||
call_kwargs = ctx["agent"].execute_task.call_args
|
||||
assert call_kwargs.kwargs.get("skip_permissions") is True
|
||||
|
||||
def test_phase1_saves_outline_file(self, tmp_db, tmp_path):
|
||||
ctx = self._make_ctx(tmp_db, tmp_path)
|
||||
create_content(
|
||||
url="https://example.com",
|
||||
keyword="plumbing services",
|
||||
ctx=ctx,
|
||||
)
|
||||
# The outline should have been saved
|
||||
outline_dir = tmp_path / "outlines" / "plumbing-services"
|
||||
assert outline_dir.exists()
|
||||
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):
|
||||
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
|
||||
|
||||
def test_phase1_includes_clickup_sync_marker(self, tmp_db, tmp_path):
|
||||
ctx = self._make_ctx(tmp_db, tmp_path)
|
||||
result = create_content(
|
||||
url="https://example.com",
|
||||
keyword="test keyword",
|
||||
ctx=ctx,
|
||||
)
|
||||
assert "## ClickUp Sync" in result
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# create_content — Phase 2
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestCreateContentPhase2:
|
||||
def _setup_phase2(self, tmp_db, tmp_path):
|
||||
"""Set up an outline_review state and outline file, return ctx."""
|
||||
cfg = Config()
|
||||
cfg.content = ContentConfig(outline_dir=str(tmp_path / "outlines"))
|
||||
|
||||
# Create the outline file
|
||||
outline_dir = tmp_path / "outlines" / "plumbing-services"
|
||||
outline_dir.mkdir(parents=True)
|
||||
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 {
|
||||
"agent": agent,
|
||||
"config": cfg,
|
||||
"db": tmp_db,
|
||||
"clickup_task_id": "task456",
|
||||
}
|
||||
|
||||
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",
|
||||
ctx=ctx,
|
||||
)
|
||||
assert "Phase 2 Complete" in result
|
||||
|
||||
def test_phase2_reads_outline(self, tmp_db, tmp_path):
|
||||
ctx = self._setup_phase2(tmp_db, tmp_path)
|
||||
create_content(
|
||||
url="https://example.com/plumbing",
|
||||
keyword="plumbing services",
|
||||
ctx=ctx,
|
||||
)
|
||||
call_args = ctx["agent"].execute_task.call_args
|
||||
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)
|
||||
create_content(
|
||||
url="https://example.com/plumbing",
|
||||
keyword="plumbing services",
|
||||
ctx=ctx,
|
||||
)
|
||||
content_file = tmp_path / "outlines" / "plumbing-services" / "final-content.md"
|
||||
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)
|
||||
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
|
||||
|
||||
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",
|
||||
ctx=ctx,
|
||||
)
|
||||
assert "## ClickUp Sync" in result
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# continue_content
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestContinueContent:
|
||||
def test_requires_keyword(self, tmp_db):
|
||||
ctx = {"agent": MagicMock(), "db": tmp_db, "config": Config()}
|
||||
assert continue_content(keyword="", ctx=ctx).startswith("Error:")
|
||||
|
||||
def test_no_matching_entry(self, tmp_db):
|
||||
ctx = {"agent": MagicMock(), "db": tmp_db, "config": Config()}
|
||||
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):
|
||||
cfg = Config()
|
||||
cfg.content = ContentConfig(outline_dir=str(tmp_path / "outlines"))
|
||||
|
||||
# Create outline file
|
||||
outline_dir = tmp_path / "outlines" / "plumbing-services"
|
||||
outline_dir.mkdir(parents=True)
|
||||
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": "",
|
||||
}
|
||||
tmp_db.kv_set("clickup:task:task789:state", json.dumps(state))
|
||||
|
||||
agent = MagicMock()
|
||||
agent.execute_task.return_value = "# Full content"
|
||||
ctx = {"agent": agent, "db": tmp_db, "config": cfg}
|
||||
result = continue_content(keyword="plumbing services", ctx=ctx)
|
||||
assert "Phase 2 Complete" in result
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Error propagation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestErrorPropagation:
|
||||
def test_phase1_execution_error_sets_failed_state(self, tmp_db, tmp_path):
|
||||
cfg = Config()
|
||||
cfg.content = ContentConfig(outline_dir=str(tmp_path / "outlines"))
|
||||
agent = MagicMock()
|
||||
agent.execute_task.side_effect = RuntimeError("CLI crashed")
|
||||
ctx = {
|
||||
"agent": agent,
|
||||
"config": cfg,
|
||||
"db": tmp_db,
|
||||
"clickup_task_id": "task_err",
|
||||
}
|
||||
result = create_content(
|
||||
url="https://example.com",
|
||||
keyword="test",
|
||||
ctx=ctx,
|
||||
)
|
||||
assert "Error:" in result
|
||||
raw = tmp_db.kv_get("clickup:task:task_err:state")
|
||||
state = json.loads(raw)
|
||||
assert state["state"] == "failed"
|
||||
|
||||
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()
|
||||
agent.execute_task.return_value = "Error: something went wrong"
|
||||
ctx = {
|
||||
"agent": agent,
|
||||
"config": cfg,
|
||||
"db": tmp_db,
|
||||
"clickup_task_id": "task_err2",
|
||||
}
|
||||
result = create_content(
|
||||
url="https://example.com",
|
||||
keyword="test",
|
||||
ctx=ctx,
|
||||
)
|
||||
assert result.startswith("Error:")
|
||||
raw = tmp_db.kv_get("clickup:task:task_err2:state")
|
||||
state = json.loads(raw)
|
||||
assert state["state"] == "failed"
|
||||
Loading…
Reference in New Issue