Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions .env.example
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,13 @@ MNEMONIK_DRY_RUN=true
# Optional: local protocol-facts file used when the Mnemonik MCP is unavailable.
# MNEMONIK_FACTS_PATH=./facts.yaml

# --- Quality gate (for `post --from-file` / `score`) -----------------------
# A finished article is scored by claude-blog's analyzer and must reach this
# score (0-100) before it can be published. The gate fails CLOSED: if the path
# below is unset or the analyzer errors, nothing publishes.
# MNEMONIK_CLAUDE_BLOG_PATH=../claude-blog
# MNEMONIK_MIN_SCORE=80

# --- Telegram channel ------------------------------------------------------
# Bot must be an ADMIN of the target channel.
TELEGRAM_BOT_TOKEN=
Expand Down
25 changes: 23 additions & 2 deletions AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,29 @@ attestation of what went out.
- `src/mnemonik_blogger/grounding/mnemonik.py` — `Grounding` protocol; static /
file fallbacks; wire the live Mnemonik MCP here.
- `src/mnemonik_blogger/publish/` — one adapter per platform + a registry.
- `src/mnemonik_blogger/agent.py` — ground → generate → render → publish → attest.
- `src/mnemonik_blogger/cli.py` — `platforms`, `preview`, `post`.
- `src/mnemonik_blogger/agent.py` — ground → generate → render → publish → attest;
plus `run_campaign_from_article` (ingest → quality gate → publish).
- `src/mnemonik_blogger/content/ingest.py` — parse a finished Markdown article
(front-matter + body) into a `SourcePost`.
- `src/mnemonik_blogger/quality.py` — score an article via claude-blog's
`analyze_blog.py` and gate publishing on `score.total >= min_score`. Fails closed.
- `src/mnemonik_blogger/cli.py` — `platforms`, `preview`, `post` (`--from-file`), `score`.

## Write → gate → publish

For finished articles (e.g. written by claude-blog), publishing is **quality-gated**:

```bash
# Score only (no publish); exits non-zero below threshold.
MNEMONIK_CLAUDE_BLOG_PATH=../claude-blog mnemonik-blogger score article.md

# Publish only if score.total >= min_score (default 80); dry-run unless --live.
MNEMONIK_CLAUDE_BLOG_PATH=../claude-blog \
mnemonik-blogger post --from-file article.md --platforms auto
```

The gate **fails closed**: if `MNEMONIK_CLAUDE_BLOG_PATH` is unset or the analyzer
errors, the article is blocked and nothing publishes.

## Rules

Expand Down
62 changes: 58 additions & 4 deletions src/mnemonik_blogger/agent.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,25 +6,38 @@
from __future__ import annotations

from dataclasses import dataclass, field
from pathlib import Path

from .config import Platform, Settings
from .content.generate import build_source_post, render_all
from .content.ingest import ingest_article
from .grounding.mnemonik import Grounding, default_grounding
from .models import PublishResult, SourcePost
from .publish import build_publisher
from .quality import QualityReport, score_article


@dataclass
class CampaignResult:
post: SourcePost
results: list[PublishResult] = field(default_factory=list)
# Set for article-based campaigns; None for direct brief-based campaigns.
quality: QualityReport | None = None
# True when the quality gate blocked publishing (results is then empty).
blocked: bool = False

@property
def ok(self) -> bool:
return all(r.ok for r in self.results)
return not self.blocked and all(r.ok for r in self.results)

def summary(self) -> str:
lines = [f"# {self.post.title}"]
if self.quality is not None:
lines.append(self.quality.summary())
if self.blocked:
lines.append("BLOCKED: quality gate not met - nothing was published.")
for issue in self.quality.issues if self.quality else []:
lines.append(f" - {issue}")
for r in self.results:
status = "DRY-RUN" if r.dry_run else ("OK" if r.ok else "FAIL")
extra = r.primary_url or r.error or ""
Expand All @@ -47,8 +60,49 @@ def run_campaign(
grounding = grounding or default_grounding(settings.facts_path)

post = build_source_post(title=title, brief=brief, link=link, tags=tags, grounding=grounding)
rendered = render_all(post, platforms)
results = _publish_post(post, platforms, settings, grounding, attest=attest)
return CampaignResult(post=post, results=results)


def run_campaign_from_article(
*,
article: str | Path,
platforms: list[Platform],
settings: Settings,
grounding: Grounding | None = None,
attest: bool = True,
min_score: int | None = None,
) -> CampaignResult:
"""Publish a finished Markdown article, but only if it clears the quality gate.

The article is scored by claude-blog's analyzer. If ``score.total`` is below
the threshold (or the gate cannot be evaluated), nothing is published and the
result is marked ``blocked``. The gate fails closed.
"""
threshold = settings.min_score if min_score is None else min_score
report = score_article(article, claude_blog_path=settings.claude_blog_path, threshold=threshold)

# ingest_article enforces guardrails; do it before any publish.
post = ingest_article(article)

if not report.passed:
return CampaignResult(post=post, results=[], quality=report, blocked=True)

grounding = grounding or default_grounding(settings.facts_path)
results = _publish_post(post, platforms, settings, grounding, attest=attest)
return CampaignResult(post=post, results=results, quality=report)


def _publish_post(
post: SourcePost,
platforms: list[Platform],
settings: Settings,
grounding: Grounding,
*,
attest: bool,
) -> list[PublishResult]:
"""Render and publish a post to each platform, then optionally attest."""
rendered = render_all(post, platforms)
results: list[PublishResult] = []
for platform in platforms:
publisher = build_publisher(platform, settings)
Expand All @@ -57,9 +111,9 @@ def run_campaign(
if attest and not settings.dry_run:
published = [r for r in results if r.ok and not r.dry_run]
if published:
record = f"Published '{title}' to: " + ", ".join(
record = f"Published '{post.title}' to: " + ", ".join(
f"{r.platform.value} ({r.primary_url or ', '.join(r.ids)})" for r in published
)
grounding.attest(record, tags=["mnemonik-blogger", "published"])

return CampaignResult(post=post, results=results)
return results
62 changes: 50 additions & 12 deletions src/mnemonik_blogger/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,11 +12,12 @@

import typer

from .agent import run_campaign
from .agent import run_campaign, run_campaign_from_article
from .config import Platform, Settings, load_settings
from .content.formatters import render
from .content.generate import build_source_post
from .grounding.mnemonik import default_grounding
from .quality import score_article

app = typer.Typer(add_completion=False, help="Mnemonik multi-platform blogger agent.")

Expand Down Expand Up @@ -70,30 +71,67 @@ def preview(

@app.command()
def post(
title: str = typer.Option(..., help="Post title."),
brief: str = typer.Option(..., help="The angle / body of the post."),
title: str = typer.Option(None, help="Post title (with --brief)."),
brief: str = typer.Option(None, help="The angle / body of the post (with --title)."),
from_file: str = typer.Option(
None,
"--from-file",
help="Publish a finished Markdown article instead of a brief (quality-gated).",
),
link: str = typer.Option(None, help="Canonical URL to include."),
tags: str = typer.Option(None, help="Comma-separated hashtags."),
target: str = typer.Option("auto", "--platforms", help="Comma list or 'auto'."),
live: bool = typer.Option(False, "--live", help="Actually publish (overrides dry-run)."),
min_score: int = typer.Option(
None, "--min-score", help="Quality threshold for --from-file (default: config)."
),
) -> None:
"""Generate and publish a post across platforms."""
"""Publish from a brief (--title/--brief) or a finished article (--from-file)."""
settings = load_settings()
if live:
settings.dry_run = False
chosen = _parse_platforms(target, settings)
result = run_campaign(
title=title,
brief=brief,
link=link,
tags=[t.strip() for t in tags.split(",")] if tags else None,
platforms=chosen,
settings=settings,
)

if from_file:
result = run_campaign_from_article(
article=from_file,
platforms=chosen,
settings=settings,
min_score=min_score,
)
else:
if not title or not brief:
raise typer.BadParameter("provide --title and --brief, or --from-file")
result = run_campaign(
title=title,
brief=brief,
link=link,
tags=[t.strip() for t in tags.split(",")] if tags else None,
platforms=chosen,
settings=settings,
)
typer.echo(result.summary())
if not result.ok:
raise typer.Exit(code=1)


@app.command()
def score(
article: str = typer.Argument(..., help="Path to a finished Markdown article."),
min_score: int = typer.Option(
None, "--min-score", help="Threshold to gate against (default: config)."
),
) -> None:
"""Score an article with claude-blog's analyzer (no publishing)."""
settings = load_settings()
threshold = settings.min_score if min_score is None else min_score
report = score_article(article, claude_blog_path=settings.claude_blog_path, threshold=threshold)
typer.echo(report.summary())
for issue in report.issues:
typer.echo(f" - {issue}")
if not report.passed:
raise typer.Exit(code=1)


if __name__ == "__main__":
app()
6 changes: 6 additions & 0 deletions src/mnemonik_blogger/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -108,6 +108,12 @@ class Settings(_Base):
dry_run: bool = True
# Path to a local protocol-facts file used when the Mnemonik MCP is unavailable.
facts_path: str | None = None
# Quality gate: a finished article must score >= min_score (claude-blog's
# analyzer) before --from-file publishing is allowed. 0-100.
min_score: int = 80
# Path to a claude-blog checkout; its scripts/analyze_blog.py scores articles.
# Required for --from-file; absence fails the gate closed (nothing publishes).
claude_blog_path: str | None = None

telegram: TelegramConfig = Field(default_factory=TelegramConfig)
discord: DiscordConfig = Field(default_factory=DiscordConfig)
Expand Down
5 changes: 3 additions & 2 deletions src/mnemonik_blogger/content/generate.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,8 @@
from . import formatters, voice


def _enforce_guardrails(text: str) -> None:
def enforce_guardrails(text: str) -> None:
"""Raise if `text` contains a banned phrase. Used by both brief- and file-based flows."""
lowered = text.lower()
for banned in voice.BANNED_PHRASES:
if banned in lowered:
Expand Down Expand Up @@ -44,7 +45,7 @@ def build_source_post(
proof = " ".join(f.text for f in facts[:2])
body = f"{body}\n\nWhy it holds up: {proof}"

_enforce_guardrails(title + " " + body)
enforce_guardrails(title + " " + body)

return SourcePost(
title=title,
Expand Down
86 changes: 86 additions & 0 deletions src/mnemonik_blogger/content/ingest.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,86 @@
"""Ingest a finished article (Markdown + optional YAML front-matter) into a SourcePost.

claude-blog produces articles as Markdown files, optionally led by a YAML
front-matter block delimited by ``---``. This module turns such a file into the
blogger's canonical :class:`SourcePost` so the publish pipeline can repurpose it
per platform.

Resolution rules:
* title - front-matter ``title``; else the first H1; else the file stem.
* summary - front-matter ``summary``/``description``; else the first paragraph.
* tags - front-matter ``tags`` (list or comma string).
* link - front-matter ``link``/``canonical``/``url``.

The article is treated as already-written and reviewed, so its body is not
re-grounded or mutated; only banned-phrase guardrails are enforced.
"""

from __future__ import annotations

import re
from pathlib import Path
from typing import Any

import yaml

from ..models import SourcePost
from .generate import enforce_guardrails

_FRONTMATTER = re.compile(r"^---\s*\n(.*?)\n---\s*\n", re.DOTALL)
_H1 = re.compile(r"^#\s+(.+?)\s*$", re.MULTILINE)


def _split_frontmatter(text: str) -> tuple[dict[str, Any], str]:
match = _FRONTMATTER.match(text)
if not match:
return {}, text
parsed = yaml.safe_load(match.group(1))
meta: dict[str, Any] = parsed if isinstance(parsed, dict) else {}
return meta, text[match.end() :]


def _first_paragraph(body: str) -> str:
for block in body.split("\n\n"):
stripped = block.strip()
if stripped and not stripped.startswith("#"):
return " ".join(stripped.split())
return ""


def _coerce_tags(raw: Any) -> list[str]:
if isinstance(raw, str):
parts = raw.split(",")
elif isinstance(raw, list):
parts = [str(t) for t in raw]
else:
return []
return [t.strip() for t in parts if t.strip()]


def ingest_article(path: str | Path) -> SourcePost:
"""Parse a Markdown article file into a SourcePost (raises on guardrail violation)."""
path = Path(path)
meta, body = _split_frontmatter(path.read_text(encoding="utf-8"))
body = body.strip()

fm_title = meta.get("title")
if fm_title:
title = str(fm_title)
else:
h1 = _H1.search(body)
title = h1.group(1).strip() if h1 else path.stem
# Drop the leading H1 so it isn't duplicated in the social copy.
body = _H1.sub("", body, count=1).strip()

summary = meta.get("summary") or meta.get("description") or _first_paragraph(body)
link = meta.get("link") or meta.get("canonical") or meta.get("url")

enforce_guardrails(f"{title} {body}")

return SourcePost(
title=title,
body=body,
summary=str(summary)[:200],
tags=_coerce_tags(meta.get("tags")),
link=str(link) if link else None,
)
Loading
Loading