feat(cmo): hydrate approved queue with generated copy and executable x-cli commands

ccross2 · ccross2 · commit 877550614e92 · 2026-04-02T14:20:34.000-04:00
diff --git a/ops/cmo-automation/README.md b/ops/cmo-automation/README.md
@@ -12,7 +12,8 @@ What this does today
 2) Analyze behavior and output baseline metrics
 3) Generate a conservative dry-run engagement queue
 4) Score the queue for risk/quality before any execution
-5) Build execution report from approved actions (dry-run default)
+5) Hydrate approved actions with copy + x-cli commands
+6) Build execution report from approved actions (dry-run default)
 
 Current mode
 - Dry-run / assisted only.
@@ -26,6 +27,7 @@ Structure
 - scripts/generate_engagement_queue.py
 - scripts/review_engagement_queue.py
 - scripts/reconstruct_fiverr_playbook.py
+- scripts/hydrate_approved_queue.py
 - scripts/execute_approved_queue.py
 - reports/CMO-AUTOMATION-IMPLEMENTATION-PLAN.md
 - reports/CMO-CUTOVER-48H-RUNBOOK.md
@@ -40,7 +42,8 @@ Quick start
   - python3 scripts/generate_engagement_queue.py
   - python3 scripts/review_engagement_queue.py
   - python3 scripts/reconstruct_fiverr_playbook.py
-  - python3 scripts/execute_approved_queue.py
+  - python3 scripts/hydrate_approved_queue.py
+  - python3 scripts/execute_approved_queue.py --input reports/cmo-hydrated-queue.json
 
 Outputs
 - data/latest.json
@@ -49,6 +52,8 @@ Outputs
 - data/engagement-queue.json
 - reports/cmo-queue-review.json
 - reports/cmo-queue-review.md
+- reports/cmo-hydrated-queue.json
+- reports/cmo-hydrated-queue.md
 - reports/cmo-execution-report.json
 - reports/cmo-execution-report.md
 
diff --git a/ops/cmo-automation/scripts/hydrate_approved_queue.py b/ops/cmo-automation/scripts/hydrate_approved_queue.py
@@ -0,0 +1,244 @@
+#!/usr/bin/env python3
+import argparse
+import json
+import os
+import re
+import subprocess
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Callable
+
+ROOT = Path(__file__).resolve().parents[1]
+DEFAULT_REVIEW = ROOT / "reports" / "cmo-queue-review.json"
+POLICY = ROOT / "config" / "operating_policy.json"
+OUT_JSON = ROOT / "reports" / "cmo-hydrated-queue.json"
+OUT_MD = ROOT / "reports" / "cmo-hydrated-queue.md"
+
+Resolver = Callable[[str | None, str], dict | None]
+
+
+def load_json(path: Path):
+    if not path.exists():
+        raise SystemExit(f"Missing required input: {path}")
+    return json.loads(path.read_text())
+
+
+def load_credential_mapping() -> None:
+    mapping = {
+        "X_API_KEY": os.getenv("X_API_KEY") or os.getenv("TWITTER_API_KEY"),
+        "X_API_SECRET": os.getenv("X_API_SECRET") or os.getenv("TWITTER_API_SECRET"),
+        "X_BEARER_TOKEN": os.getenv("X_BEARER_TOKEN") or os.getenv("TWITTER_BEARER_TOKEN"),
+        "X_ACCESS_TOKEN": os.getenv("X_ACCESS_TOKEN") or os.getenv("TWITTER_ACCESS_TOKEN"),
+        "X_ACCESS_TOKEN_SECRET": os.getenv("X_ACCESS_TOKEN_SECRET") or os.getenv("TWITTER_ACCESS_SECRET"),
+    }
+    missing = [k for k, v in mapping.items() if not v]
+    if missing:
+        raise SystemExit(f"Missing required credentials: {', '.join(missing)}")
+    os.environ.update(mapping)
+
+
+def run_json(cmd: list[str]):
+    p = subprocess.run(cmd, capture_output=True, text=True)
+    if p.returncode != 0:
+        raise RuntimeError(f"Command failed: {' '.join(cmd)}\nSTDERR:\n{p.stderr}")
+    return json.loads(p.stdout)
+
+
+def normalize_text(text: str, limit: int = 100) -> str:
+    clean = re.sub(r"\s+", " ", (text or "")).strip()
+    if len(clean) <= limit:
+        return clean
+    return clean[: limit - 1].rstrip() + "…"
+
+
+def build_root_text(account: str, role: str) -> str:
+    ideas = {
+        "founder": "Building AI products is mostly distribution math + feedback loops. Own both, ship faster.",
+        "brand": "Good AI products win when onboarding is instant, outcomes are measurable, and support feels human.",
+        "product-agent": "Agent workflows improve when memory, orchestration, and evals are designed as one system.",
+    }
+    base = ideas.get(role, ideas["brand"])
+    return f"{base} #{account}"[:278]
+
+
+def safe_founder_reply(target_user: str, excerpt: str) -> str:
+    return (
+        f"@{target_user} Good signal here. I care less about hype and more about repeatable distribution + retention. "
+        f"{excerpt}"
+    )[:278]
+
+
+def build_reply_text(account: str, role: str, target_user: str, source_text: str) -> str:
+    excerpt = normalize_text(source_text, limit=80)
+    if role == "founder":
+        return safe_founder_reply(target_user, excerpt)
+    if role == "product-agent":
+        return (
+            f"@{target_user} Nice thread. Curious what your eval loop looks like once this is in production. "
+            f"{excerpt}"
+        )[:278]
+    return (
+        f"@{target_user} Solid point. We see the same pattern in shipping: clear UX + measurable outcomes compound. "
+        f"{excerpt}"
+    )[:278]
+
+
+def contains_denylist(text: str, keywords: list[str]) -> bool:
+    t = (text or "").lower()
+    return any(k.lower() in t for k in keywords)
+
+
+def default_resolver(target_user: str | None, account: str) -> dict | None:
+    account_queries = {
+        "TheCesarCross": "AI founders OR agentic workflow OR product distribution",
+        "sovren_software": "AI automation OR workflow systems OR developer tools",
+        "mrhaven_agent": "agent memory OR orchestration OR evals",
+    }
+
+    if target_user:
+        query = f"from:{target_user} -is:retweet"
+    else:
+        query = account_queries.get(account, "AI products OR automation")
+
+    data = run_json(["x-cli", "-j", "tweet", "search", query, "--max", "10"])
+    if not isinstance(data, list) or not data:
+        return None
+
+    for t in data:
+        if not isinstance(t, dict):
+            continue
+        tweet_id = t.get("id")
+        text = t.get("text")
+        author = ((t.get("author") or {}).get("username") if isinstance(t.get("author"), dict) else None)
+        if tweet_id and text:
+            return {"id": str(tweet_id), "text": text, "author": author}
+    return None
+
+
+def hydrate_single_action(action: dict, policy: dict, resolver: Resolver) -> dict:
+    out = dict(action)
+    account = out.get("account", "")
+    role = policy.get("account_strategy", {}).get(account, {}).get("role", "brand")
+    founder_keywords = policy.get("founder_denylist", {}).get("keywords", [])
+
+    if out.get("action") == "root_post":
+        post_text = build_root_text(account, role)
+        if role == "founder" and contains_denylist(post_text, founder_keywords):
+            out["hydration_status"] = "blocked"
+            out["hydration_reason"] = "founder_denylist_hit"
+            return out
+
+        out["post_text"] = post_text
+        out["x_cli_command"] = ["x-cli", "-j", "tweet", "post", post_text]
+        out["hydration_status"] = "hydrated"
+        return out
+
+    if out.get("action") == "reply":
+        candidate = resolver(out.get("target_user"), account)
+        if not candidate:
+            out["hydration_status"] = "blocked"
+            out["hydration_reason"] = "no_candidate_tweet"
+            return out
+
+        target_user = out.get("target_user") or candidate.get("author") or "builder"
+        reply_text = build_reply_text(account, role, target_user, candidate.get("text", ""))
+
+        if role == "founder" and contains_denylist(reply_text, founder_keywords):
+            out["hydration_status"] = "blocked"
+            out["hydration_reason"] = "founder_denylist_hit"
+            return out
+
+        tweet_id = str(candidate["id"])
+        out["target_user"] = target_user
+        out["target_tweet_id"] = tweet_id
+        out["reply_text"] = reply_text
+        out["execution_mode"] = "quote_workaround"
+        out["x_cli_command"] = ["x-cli", "-j", "tweet", "quote", tweet_id, reply_text]
+        out["hydration_status"] = "hydrated"
+        return out
+
+    out["hydration_status"] = "blocked"
+    out["hydration_reason"] = "unsupported_action_type"
+    return out
+
+
+def hydrate_review(review: dict, policy: dict, resolver: Resolver) -> dict:
+    hydrated = dict(review)
+    hydrated_accounts = {}
+
+    total = 0
+    ready = 0
+    blocked = 0
+
+    for account, payload in review.get("accounts", {}).items():
+        approved = payload.get("approved_actions", [])
+        hydrated_approved = []
+        for action in approved:
+            total += 1
+            h = hydrate_single_action(action, policy, resolver)
+            hydrated_approved.append(h)
+            if h.get("hydration_status") == "hydrated":
+                ready += 1
+            else:
+                blocked += 1
+
+        new_payload = dict(payload)
+        new_payload["approved_actions"] = hydrated_approved
+        hydrated_accounts[account] = new_payload
+
+    hydrated["accounts"] = hydrated_accounts
+    hydrated["hydrated_at"] = datetime.now(timezone.utc).isoformat()
+    hydrated["hydration_summary"] = {
+        "total_approved": total,
+        "hydrated": ready,
+        "blocked": blocked,
+    }
+    return hydrated
+
+
+def write_reports(hydrated: dict) -> None:
+    OUT_JSON.write_text(json.dumps(hydrated, indent=2))
+
+    s = hydrated.get("hydration_summary", {})
+    lines = [
+        "# CMO Hydrated Queue",
+        f"Hydrated at: {hydrated.get('hydrated_at')}",
+        f"- total approved: {s.get('total_approved', 0)}",
+        f"- hydrated: {s.get('hydrated', 0)}",
+        f"- blocked: {s.get('blocked', 0)}",
+        "",
+    ]
+
+    for account, payload in hydrated.get("accounts", {}).items():
+        actions = payload.get("approved_actions", [])
+        ok = sum(1 for a in actions if a.get("hydration_status") == "hydrated")
+        no = len(actions) - ok
+        lines.append(f"## @{account} hydrated={ok} blocked={no}")
+        for a in actions:
+            lines.append(
+                f"- {a.get('action')} target={a.get('target_user')} status={a.get('hydration_status')} reason={a.get('hydration_reason')}"
+            )
+        lines.append("")
+
+    OUT_MD.write_text("\n".join(lines))
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Hydrate approved queue actions with copy and executable x-cli commands")
+    parser.add_argument("--input", default=str(DEFAULT_REVIEW), help="Path to cmo-queue-review.json")
+    return parser.parse_args()
+
+
+def main() -> None:
+    args = parse_args()
+    load_credential_mapping()
+    review = load_json(Path(args.input))
+    policy = load_json(POLICY)
+    hydrated = hydrate_review(review, policy, default_resolver)
+    write_reports(hydrated)
+    print(str(OUT_JSON))
+    print(str(OUT_MD))
+
+
+if __name__ == "__main__":
+    main()
diff --git a/ops/cmo-automation/tests/test_hydrate_approved_queue.py b/ops/cmo-automation/tests/test_hydrate_approved_queue.py
@@ -0,0 +1,57 @@
+import sys
+import unittest
+from pathlib import Path
+
+SCRIPTS = Path(__file__).resolve().parents[1] / "scripts"
+sys.path.insert(0, str(SCRIPTS))
+
+import hydrate_approved_queue  # noqa: E402
+
+
+class HydrateApprovedQueueTests(unittest.TestCase):
+    def test_hydrate_root_post_adds_text_and_command(self):
+        action = {"account": "sovren_software", "action": "root_post", "priority": "high"}
+        policy = {
+            "account_strategy": {"sovren_software": {"role": "brand"}},
+            "founder_denylist": {"keywords": []},
+        }
+
+        out = hydrate_approved_queue.hydrate_single_action(action, policy, resolver=lambda *_: None)
+
+        self.assertEqual("hydrated", out["hydration_status"])
+        self.assertTrue(out["post_text"])
+        self.assertEqual("x-cli", out["x_cli_command"][0])
+        self.assertEqual("post", out["x_cli_command"][3])
+
+    def test_hydrate_reply_uses_quote_workaround_with_candidate(self):
+        action = {"account": "TheCesarCross", "action": "reply", "target_user": "alice"}
+        policy = {
+            "account_strategy": {"TheCesarCross": {"role": "founder"}},
+            "founder_denylist": {"keywords": ["giveaway"]},
+        }
+
+        def resolver(target_user, account):
+            return {"id": "12345", "text": "Shipping useful agent products this week."}
+
+        out = hydrate_approved_queue.hydrate_single_action(action, policy, resolver=resolver)
+
+        self.assertEqual("hydrated", out["hydration_status"])
+        self.assertEqual("12345", out["target_tweet_id"])
+        self.assertEqual(["x-cli", "-j", "tweet", "quote"], out["x_cli_command"][0:4])
+        self.assertIn("@alice", out["reply_text"])
+
+    def test_hydrate_reply_blocks_without_candidate(self):
+        action = {"account": "mrhaven_agent", "action": "reply", "target_user": "nobody"}
+        policy = {
+            "account_strategy": {"mrhaven_agent": {"role": "product-agent"}},
+            "founder_denylist": {"keywords": []},
+        }
+
+        out = hydrate_approved_queue.hydrate_single_action(action, policy, resolver=lambda *_: None)
+
+        self.assertEqual("blocked", out["hydration_status"])
+        self.assertEqual("no_candidate_tweet", out["hydration_reason"])
+
+
+if __name__ == "__main__":
+    unittest.main()
