feat(cmo): add approved-queue executor with dry-run execution report

Author: ccross2

ops/cmo-automation/README.md

@@ -12,6 +12,7 @@ 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)
 
 Current mode
 - Dry-run / assisted only.
@@ -25,6 +26,7 @@ Structure
 - scripts/generate_engagement_queue.py
 - scripts/review_engagement_queue.py
 - scripts/reconstruct_fiverr_playbook.py
+- scripts/execute_approved_queue.py
 - reports/CMO-AUTOMATION-IMPLEMENTATION-PLAN.md
 - reports/CMO-CUTOVER-48H-RUNBOOK.md
 
@@ -38,6 +40,7 @@ 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
 
 Outputs
 - data/latest.json
@@ -46,6 +49,8 @@ Outputs
 - data/engagement-queue.json
 - reports/cmo-queue-review.json
 - reports/cmo-queue-review.md
+- reports/cmo-execution-report.json
+- reports/cmo-execution-report.md
 
 Guardrails
 - Generic short-reply behavior is penalized.

ops/cmo-automation/scripts/execute_approved_queue.py

@@ -0,0 +1,170 @@
+#!/usr/bin/env python3
+import argparse
+import json
+import os
+import subprocess
+from datetime import datetime, timezone
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+DEFAULT_INPUT = ROOT / "reports" / "cmo-queue-review.json"
+OUT_JSON = ROOT / "reports" / "cmo-execution-report.json"
+OUT_MD = ROOT / "reports" / "cmo-execution-report.md"
+
+
+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 flatten_approved_actions(review: dict) -> list[dict]:
+    actions: list[dict] = []
+    for account, payload in review.get("accounts", {}).items():
+        for action in payload.get("approved_actions", []):
+            if not action.get("account"):
+                action = {**action, "account": account}
+            actions.append(action)
+    return actions
+
+
+def classify_action(action: dict) -> tuple[bool, str]:
+    if isinstance(action.get("x_cli_command"), list) and action["x_cli_command"]:
+        return True, "explicit_command"
+
+    kind = action.get("action")
+    if kind == "root_post":
+        text = action.get("post_text") or action.get("text")
+        if not text:
+            return False, "missing_root_text"
+        return False, "missing_x_cli_command"
+
+    if kind == "reply":
+        if not action.get("reply_to_tweet_id") or not (action.get("reply_text") or action.get("text")):
+            return False, "missing_reply_payload"
+        return False, "missing_x_cli_command"
+
+    return False, "unsupported_action_type"
+
+
+def run_live(action: dict) -> dict:
+    cmd = action.get("x_cli_command")
+    if not isinstance(cmd, list) or not cmd:
+        return {"status": "blocked", "reason": "missing_x_cli_command"}
+
+    proc = subprocess.run(cmd, capture_output=True, text=True)
+    return {
+        "status": "executed" if proc.returncode == 0 else "failed",
+        "returncode": proc.returncode,
+        "stdout": proc.stdout.strip(),
+        "stderr": proc.stderr.strip(),
+    }
+
+
+def build_execution_plan(review: dict, live: bool = False) -> dict:
+    actions = flatten_approved_actions(review)
+    result_actions = []
+
+    for idx, action in enumerate(actions, start=1):
+        ready, reason = classify_action(action)
+        row = {
+            "id": idx,
+            "account": action.get("account"),
+            "action": action.get("action"),
+            "target_user": action.get("target_user"),
+            "ready": ready,
+            "reason": reason,
+            "source": action,
+        }
+        if live and ready:
+            row.update(run_live(action))
+        elif live:
+            row.update({"status": "blocked"})
+        else:
+            row.update({"status": "planned" if ready else "blocked"})
+        result_actions.append(row)
+
+    ready_count = sum(1 for a in result_actions if a["ready"])
+    executed_count = sum(1 for a in result_actions if a.get("status") == "executed")
+    failed_count = sum(1 for a in result_actions if a.get("status") == "failed")
+
+    plan = {
+        "generated_at": datetime.now(timezone.utc).isoformat(),
+        "mode": "live" if live else "dry-run",
+        "input_review_mode": review.get("mode", "unknown"),
+        "summary": {
+            "total_approved": len(actions),
+            "ready_to_execute": ready_count,
+            "blocked": len(actions) - ready_count,
+            "executed": executed_count,
+            "failed": failed_count,
+        },
+        "actions": result_actions,
+    }
+    return plan
+
+
+def write_reports(plan: dict) -> None:
+    OUT_JSON.write_text(json.dumps(plan, indent=2))
+
+    lines = [
+        "# CMO Approved Queue Execution Report",
+        f"Generated: {plan['generated_at']}",
+        f"Mode: {plan['mode']}",
+        f"Input review mode: {plan['input_review_mode']}",
+        "",
+        "## Summary",
+        f"- total approved: {plan['summary']['total_approved']}",
+        f"- ready to execute: {plan['summary']['ready_to_execute']}",
+        f"- blocked: {plan['summary']['blocked']}",
+        f"- executed: {plan['summary']['executed']}",
+        f"- failed: {plan['summary']['failed']}",
+        "",
+        "## Action status",
+    ]
+
+    for row in plan["actions"]:
+        lines.append(
+            f"- #{row['id']} @{row.get('account')} {row.get('action')} target={row.get('target_user')} status={row.get('status')} reason={row.get('reason')}"
+        )
+
+    OUT_MD.write_text("\n".join(lines) + "\n")
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Execute approved CMO queue actions from review output")
+    parser.add_argument("--input", default=str(DEFAULT_INPUT), help="Path to cmo-queue-review.json")
+    parser.add_argument("--live", action="store_true", help="Attempt to execute actions with explicit x_cli_command")
+    return parser.parse_args()
+
+
+def main() -> None:
+    args = parse_args()
+    review = load_json(Path(args.input))
+
+    if args.live:
+        load_credential_mapping()
+
+    plan = build_execution_plan(review, live=args.live)
+    write_reports(plan)
+
+    print(str(OUT_JSON))
+    print(str(OUT_MD))
+
+
+if __name__ == "__main__":
+    main()

ops/cmo-automation/tests/test_execute_approved_queue.py

@@ -0,0 +1,62 @@
+import sys
+import unittest
+from pathlib import Path
+
+SCRIPTS = Path(__file__).resolve().parents[1] / "scripts"
+sys.path.insert(0, str(SCRIPTS))
+
+import execute_approved_queue  # noqa: E402
+
+
+class ExecuteApprovedQueueTests(unittest.TestCase):
+    def test_flatten_approved_actions_only(self):
+        review = {
+            "accounts": {
+                "TheCesarCross": {
+                    "approved_actions": [
+                        {"account": "TheCesarCross", "action": "root_post"},
+                        {"account": "TheCesarCross", "action": "reply", "target_user": "alice"},
+                    ],
+                    "rejected_actions": [
+                        {"account": "TheCesarCross", "action": "reply", "target_user": "bob"}
+                    ],
+                },
+                "sovren_software": {
+                    "approved_actions": [
+                        {"account": "sovren_software", "action": "reply", "target_user": "carol"}
+                    ],
+                    "rejected_actions": [],
+                },
+            }
+        }
+
+        actions = execute_approved_queue.flatten_approved_actions(review)
+
+        self.assertEqual(3, len(actions))
+        self.assertEqual("root_post", actions[0]["action"])
+        self.assertEqual("carol", actions[2]["target_user"])
+
+    def test_build_execution_plan_requires_payload(self):
+        review = {
+            "mode": "dry-run",
+            "accounts": {
+                "TheCesarCross": {
+                    "approved_actions": [
+                        {"account": "TheCesarCross", "action": "root_post"},
+                        {"account": "TheCesarCross", "action": "reply", "target_user": "alice"},
+                    ],
+                }
+            },
+        }
+
+        plan = execute_approved_queue.build_execution_plan(review, live=False)
+
+        self.assertEqual(2, plan["summary"]["total_approved"])
+        self.assertEqual(0, plan["summary"]["ready_to_execute"])
+        reasons = {x["reason"] for x in plan["actions"]}
+        self.assertIn("missing_root_text", reasons)
+        self.assertIn("missing_reply_payload", reasons)
+
+
+if __name__ == "__main__":
+    unittest.main()