Building an Autoposting Pipeline with Hermes Agent: Why Waterfall Beats Parallel, and the Edge Cases Nobody Talks About

┌─────────────────────────────────────────────────────────────────────┐
│  PIPELINE: WATERFALL WITH SELECTIVE CONCURRENCY                       │
├─────────────────────────────────────────────────────────────────────┤
│                                                                     │
│  Stage 1: Content Adaptation (sequential)                           │
│    → Parse markdown source                                          │
│    → Generate platform-specific variants via Jinja2 templates       │
│    → Compress images per platform limits (FFmpeg)                    │
│                                                                     │
│  Stage 2: Primary Hub (sequential, blocking)                        │
│    → WordPress: publish full article → get canonical URL            │
│    → Dev.to: publish markdown variant → get dev.to URL              │
│                                                                     │
│  Stage 3: Social Teasers (parallel, 4 threads)                     │
│    → Bluesky: 300 chars + compressed image + link                   │
│    → Mastodon: 500 chars + media upload + link                      │
│    → Tumblr: photo post + caption + link                            │
│    → Paragraph: markdown + external image URLs + link               │
│                                                                     │
│  Stage 4: Archive (sequential)                                      │
│    → Git commit with all published URLs                             │
│    → Update LLM-Wiki index                                          │
│    → Write execution log for debugging                              │
│                                                                     │
└─────────────────────────────────────────────────────────────────────┘

# scripts/adaptation.py
import re
from pathlib import Path
from jinja2 import Environment, FileSystemLoader

TEMPLATES_DIR = Path(__file__).parent.parent / "templates"
env = Environment(loader=FileSystemLoader(TEMPLATES_DIR))

def adapt_for_platform(source_md: str, platform: str) -> str:
    """Generate platform-specific variant from canonical markdown."""
    template = env.get_template(f"{platform}.j2")

    # Platform-specific preprocessing
    if platform in ("telegram", "vk"):
        # Strip all markdown — these platforms don't render it
        text = re.sub(r'\*\*(.*?)\*\*', r'\1', source_md)   # bold
        text = re.sub(r'\*(.*?)\*', r'\1', text)             # italic
        text = re.sub(r'__(.*?)__', r'\1', text)            # underline
        text = re.sub(r'~~(.*?)~~', r'\1', text)            # strikethrough
        text = re.sub(r'\[(.*?)\]\((.*?)\)', r'\2', text)   # links → bare URL
        text = re.sub(r'!\[.*?\]\(.*?\)', '', text)        # remove images
        text = re.sub(r'#{1,6}\s+', '', text)              # remove headers
        return template.render(content=text, has_images=False)

    elif platform == "bluesky":
        # 300 graphemes hard limit — distill to hook + detail + URL
        text = re.sub(r'#{1,6}\s+', '', source_md)
        text = re.sub(r'\*\*(.*?)\*\*', r'\1', text)
        text = re.sub(r'\n+', ' ', text)
        # Leave room for URL (~30 chars)
        return text[:270].strip() + "..."

    elif platform == "devto":
        # Markdown-native but SVG doesn't render
        text = source_md.replace(".svg)", ".png)")  # rough SVG→PNG swap
        return template.render(content=text, tags=extract_tags(source_md))

    elif platform == "wordpress":
        # Full HTML conversion for WordPress REST API
        return template.render(content=markdown_to_html(source_md))

    elif platform == "mastodon":
        # 500 chars, no markdown
        text = re.sub(r'\*\*(.*?)\*\*', r'\1', source_md)
        text = re.sub(r'\n+', ' ', text)
        return text[:470].strip() + "..."

    elif platform == "paragraph":
        # Markdown accepted, but images must be external URLs
        return template.render(content=source_md, images_are_external=True)

    return source_md  # fallback

def extract_tags(md: str) -> list:
    """Extract tags from YAML frontmatter, limit to 4 for Dev.to."""
    match = re.search(r'^tags:\s*(.+)', md, re.MULTILINE)
    if match:
        tags = [t.strip() for t in match.group(1).split(",")]
        return tags[:4]  # Dev.to hard limit
    return ["python", "automation"]

# api/publishers/wordpress.py
import os
import requests

ACCESS_TOKEN = os.environ.get('WORDPRESS_ACCESS_TOKEN')
BLOG_ID = os.environ.get('WORDPRESS_BLOG_ID')

def publish_post(title: str, content: str, featured_image_url: str = None,
                 categories: list = None, tags: str = "") -> dict:
    url = f"https://public-api.wordpress.com/rest/v1.2/sites/{BLOG_ID}/posts/new"
    headers = {
        "Authorization": f"Bearer {ACCESS_TOKEN}",
        "Content-Type": "application/json",
    }
    payload = {
        "title": title,
        "content": content,
        "status": "publish",
    }
    if categories:
        payload["categories"] = categories
    if tags:
        payload["tags"] = tags
    if featured_image_url:
        payload["featured_image"] = featured_image_url

    r = requests.post(url, headers=headers, json=payload, timeout=30)
    data = r.json()

    return {
        "success": "ID" in data,
        "url": data.get("URL"),
        "id": data.get("ID")
    }

# api/publishers/devto.py
import os
import requests

API_KEY = os.environ.get('DEVTO_API_KEY')

def publish_article(title: str, body: str, tags: list, published: bool = False) -> dict:
    url = "https://dev.to/api/articles"
    headers = {
        "api-key": API_KEY,
        "Content-Type": "application/json"
    }

    payload = {
        "article": {
            "title": title,
            "body_markdown": body,
            "published": published,
            "tags": tags[:4]  # Dev.to hard limit: 4 tags max
        }
    }

    r = requests.post(url, headers=headers, json=payload, timeout=30)
    data = r.json()

    if r.status_code == 201:
        return {"success": True, "url": data.get("url"), "id": data.get("id")}
    return {"success": False, "error": data.get("error", f"HTTP {r.status_code}")}

# api/publishers/bluesky.py
import os
import requests
from datetime import datetime, timezone

HANDLE = os.environ.get('BLUESKY_HANDLE')
APP_PASSWORD = os.environ.get('BLUESKY_APP_PASSWORD')
BASE_URL = "https://bsky.social/xrpc"

def create_session() -> dict:
    r = requests.post(
        f"{BASE_URL}/com.atproto.server.createSession",
        json={"identifier": HANDLE, "password": APP_PASSWORD},
        timeout=30
    )
    data = r.json()
    return {
        "success": "accessJwt" in data,
        "accessJwt": data.get("accessJwt"),
        "did": data.get("did")
    }

def upload_blob(image_path: str, session: dict) -> dict:
    ext = image_path.lower().split(".")[-1]
    mime = {"jpg": "image/jpeg", "jpeg": "image/jpeg", "png": "image/png", "gif": "image/gif"}.get(ext, "image/png")

    with open(image_path, "rb") as f:
        data = f.read()

    r = requests.post(
        f"{BASE_URL}/com.atproto.repo.uploadBlob",
        headers={
            "Authorization": f"Bearer {session['accessJwt']}",
            "Content-Type": mime
        },
        data=data,
        timeout=120  # 1.5MB uploads need time
    )

    blob = r.json().get("blob")
    return {"success": bool(blob), "blob": blob}

def post(text: str, image_path: str = None) -> dict:
    session = create_session()
    if not session["success"]:
        return session

    now = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")

    record = {
        "$type": "app.bsky.feed.post",
        "text": text,
        "createdAt": now,
    }

    if image_path and os.path.exists(image_path):
        if os.path.getsize(image_path) > 2_000_000:
            compressed = image_path.replace(".png", "-compressed.jpg")
            os.system(f"ffmpeg -y -i {image_path} -q:v 2 {compressed}")
            image_path = compressed

        blob = upload_blob(image_path, session)
        if blob["success"]:
            record["embed"] = {
                "$type": "app.bsky.embed.images",
                "images": [{"alt": "", "image": blob["blob"]}]
            }

    r = requests.post(
        f"{BASE_URL}/com.atproto.repo.createRecord",
        headers={"Authorization": f"Bearer {session['accessJwt']}"},
        json={
            "repo": session["did"],
            "collection": "app.bsky.feed.post",
            "record": record
        },
        timeout=30
    )

    data = r.json()
    if "uri" in data:
        post_id = data["uri"].split("/")[-1]
        return {
            "success": True,
            "url": f"https://bsky.app/profile/{HANDLE}/post/{post_id}"
        }
    return {"success": False, "error": data}

# api/publishers/mastodon.py
import os
import requests

BASE_URL = os.environ.get('MASTODON_URL')
ACCESS_TOKEN = os.environ.get('MASTODON_ACCESS_TOKEN')

def upload_media(image_path: str) -> dict:
    url = f"{BASE_URL}/api/v1/media"
    headers = {"Authorization": f"Bearer {ACCESS_TOKEN}"}

    with open(image_path, "rb") as f:
        files = {"file": (os.path.basename(image_path), f)}
        r = requests.post(url, headers=headers, files=files, timeout=60)

    data = r.json()
    return {
        "success": "id" in data,
        "id": data.get("id"),
        "url": data.get("url")
    }

def post_status(text: str, media_ids: list = None) -> dict:
    url = f"{BASE_URL}/api/v1/statuses"
    headers = {"Authorization": f"Bearer {ACCESS_TOKEN}"}
    payload = {"status": text}
    if media_ids:
        payload["media_ids[]"] = media_ids

    r = requests.post(url, headers=headers, data=payload, timeout=30)
    data = r.json()

    return {
        "success": "id" in data,
        "url": data.get("url")
    }

# scripts/orchestrator.py
import sys
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor

sys.path.insert(0, str(Path(__file__).parent.parent))

from api.publishers.bluesky import post as publish_bluesky
from api.publishers.wordpress import publish_post as publish_wordpress
from api.publishers.devto import publish_article as publish_devto
from api.publishers.mastodon import post_status as publish_mastodon, upload_media as upload_mastodon_media
from api.publishers.paragraph import publish_post as publish_paragraph
from scripts.adaptation import adapt_for_platform

def run_pipeline(article_path: str, image_path: str, dry_run: bool = True):
    with open(article_path) as f:
        full_text = f.read()

    # Stage 1: Adaptation
    variants = {
        "wordpress": adapt_for_platform(full_text, "wordpress"),
        "devto": adapt_for_platform(full_text, "devto"),
        "bluesky": adapt_for_platform(full_text, "bluesky"),
        "mastodon": adapt_for_platform(full_text, "mastodon"),
        "tumblr": adapt_for_platform(full_text, "tumblr"),
        "paragraph": adapt_for_platform(full_text, "paragraph"),
    }

    if dry_run:
        print("[DRY RUN] Platform variants generated:")
        for platform, text in variants.items():
            print(f"  {platform}: {len(text)} chars")
        return

    # Stage 2: Primary Hub (sequential, blocking)
    wp = publish_wordpress({
        "title": "Article Title",
        "content": variants["wordpress"],
        "status": "publish",
        "categories": ["Productivity", "Tools"],
        "tags": "automation, python, publishing, hermes"
    })
    canonical_url = wp.get("url")

    dev = publish_devto({
        "title": "Article Title",
        "body_markdown": variants["devto"],
        "published": False,  # Draft for review
        "tags": ["python", "automation", "publishing", "hermes"]
    })

    # Stage 3: Social Teasers (parallel, 4 threads)
    def publish_bluesky_trailer():
        text = f"{variants['bluesky']} → {canonical_url}"
        return publish_bluesky(text, image_path)

    def publish_mastodon_trailer():
        media = upload_mastodon_media(image_path)
        text = f"{variants['mastodon']} → {dev.get('url', canonical_url)}"
        return publish_mastodon(text, [media["id"]] if media["success"] else [])

    def publish_tumblr_post():
        # Tumblr API: photo post with caption
        return publish_tumblr(image_path, variants["tumblr"], canonical_url)

    def publish_paragraph_article():
        return publish_paragraph("Article Title", variants["paragraph"])

    with ThreadPoolExecutor(max_workers=4) as executor:
        futures = {
            "bluesky": executor.submit(publish_bluesky_trailer),
            "mastodon": executor.submit(publish_mastodon_trailer),
            "tumblr": executor.submit(publish_tumblr_post),
            "paragraph": executor.submit(publish_paragraph_article),
        }
        social_results = {k: v.result() for k, v in futures.items()}

    # Stage 4: Archive
    log_publish_results(wp, dev, social_results)
    git_commit_with_urls(wp, dev, social_results)
    update_llm_wiki_index(wp, dev, social_results)

Bluesky: FAILED (blob > 2MB)
WordPress: SUCCESS — https://azamatsafarov.wordpress.com/...
Dev.to: SUCCESS (draft) — https://dev.to/...
Mastodon: SUCCESS — https://mastodon.social/@...