Claude Routine Stage 1 — Phase 1 Implementation Plan

class TestAuthoritativeGate(unittest.TestCase):
    """authoritative_gate computes pass from scores + hard-checks and IGNORES
    the reviewer's `pass` boolean -- unlike evaluate_scorecard, a reviewer
    saying pass=false cannot hold an otherwise-clean, on-standard article."""

    def setUp(self):
        import generate_posts as gp
        self.gp = gp

    def _card(self, hv=4, wa=4, re=4, ac=4, **extra):
        card = {"scores": {"human_voice": hv, "warmth": wa,
                           "readability": re, "accuracy": ac}}
        card.update(extra)
        return card

    def test_reviewer_pass_false_does_not_hold_a_clean_on_standard_article(self):
        # scores all clear the bar, body clean -> PASS even though reviewer said fail
        passed, flags = self.gp.authoritative_gate(
            self._card(**{"pass": False, "ai_patterns_found": ["em dash present"]}),
            "clean body without the forbidden character.")
        self.assertTrue(passed)

    def test_score_below_minimum_fails(self):
        passed, flags = self.gp.authoritative_gate(self._card(hv=2), "clean body")
        self.assertFalse(passed)
        self.assertTrue(any("human_voice=2" in str(f) for f in flags))

    def test_real_em_dash_in_body_fails(self):
        passed, flags = self.gp.authoritative_gate(self._card(), "has an — em dash")
        self.assertFalse(passed)
        self.assertIn("em_dash_in_body", flags)

    def test_fabrication_flag_fails(self):
        passed, _ = self.gp.authoritative_gate(
            self._card(flags=["fabricated statistic in section 2"]), "clean body")
        self.assertFalse(passed)

    def test_missing_score_fails(self):
        passed, _ = self.gp.authoritative_gate(self._card(hv=None), "clean body")
        self.assertFalse(passed)

def authoritative_gate(scorecard: dict, content: str) -> tuple:
    """Decide pass/fail for the internal Claude routine WITHOUT trusting the
    reviewer's self-reported `pass`. Unlike evaluate_scorecard (which seeds from
    `pass` and can only downgrade -- the asymmetry that held a clean article on a
    hallucinated em-dash veto), this starts from PASS and downgrades only on:
      - a numeric score below its REVIEW_SCORE_MINIMUMS floor,
      - a real em dash in the actual body (deterministic, not the model's count),
      - an accuracy/fabrication keyword in the flags.
    Returns (passed: bool, flags: list).
    """
    passed = True
    flags  = list(scorecard.get("flags", []))
    scores = scorecard.get("scores", {}) or {}

    for key, minimum in REVIEW_SCORE_MINIMUMS.items():
        val = scores.get(key)
        if not isinstance(val, (int, float)) or val < minimum:
            passed = False
            flags.append(f"{key}={val} below minimum {minimum}")

    if "—" in content:  # real U+2014 em dash in the body
        passed = False
        flags.append("em_dash_in_body")

    accuracy_keywords = ("fabricat", "unverif", "invent", "statistic", "percentag",
                         "specific number", "no source", "not verif", "made up",
                         "cited", "claimed", "without source")
    if flags and any(kw in " ".join(str(f) for f in flags).lower()
                     for kw in accuracy_keywords):
        passed = False

    return passed, flags

class TestSelectNextTopic(unittest.TestCase):
    def setUp(self):
        import generate_posts as gp
        self.gp = gp

    def _products(self):
        return {
            "a": {"topic": "a", "title": "A", "keyword": "ka", "format": "single_review"},
            "b": {"topic": "b", "title": "B", "keyword": "kb", "format": "roundup"},
            "c": {"topic": "c", "title": "C"},  # missing keyword/format -> skipped
        }

    def test_returns_first_unpublished_valid_topic(self):
        slug, product = self.gp.select_next_topic(self._products(), used_slugs=set())
        self.assertEqual(slug, "a")
        self.assertEqual(product["format"], "single_review")

    def test_skips_used_slugs(self):
        slug, product = self.gp.select_next_topic(self._products(), used_slugs={"a"})
        self.assertEqual(slug, "b")

    def test_skips_entries_missing_required_fields(self):
        # only 'c' left unused, but 'c' is invalid -> None
        slug_product = self.gp.select_next_topic(self._products(), used_slugs={"a", "b"})
        self.assertIsNone(slug_product)

    def test_returns_none_when_queue_empty(self):
        self.assertIsNone(self.gp.select_next_topic({}, used_slugs=set()))

def select_next_topic(products: dict, used_slugs: set):
    """Return (slug, product) for the first products.json entry that is unpublished
    and has the required fields, in dict order (same order main() consumes). Return
    None when nothing is eligible. Mirrors the topic filter + dedup in main()."""
    required = ("topic", "title", "keyword", "format")
    for slug, p in products.items():
        if slug in used_slugs:
            continue
        if all(k in p for k in required):
            return slug, p
    return None

        # TOPICS entirely from products.json -- no hardcoded list. Build the
        # capped worklist by repeatedly selecting the next eligible topic so the
        # ordering/dedup rule lives in one place (select_next_topic).
        used_slugs = build_used_slugs()
        topics = []
        remaining = dict(products)
        while len(topics) < max_articles:
            picked = select_next_topic(remaining, used_slugs)
            if picked is None:
                break
            slug, p = picked
            topics.append((p["topic"], p["title"], p["keyword"], p["format"]))
            remaining.pop(slug, None)
        log(f"Unpublished topics queued this run: {len(topics)}")

class TestBuildWriterInputs(unittest.TestCase):
    def setUp(self):
        import generate_posts as gp
        self.gp = gp

    def test_single_review_inputs_have_system_and_user(self):
        product = {"topic": "best-x", "title": "Best X", "keyword": "best x",
                   "format": "single_review", "name": "The X", "category": "dogs",
                   "species": "dog", "affiliate_url": "https://amzn.to/abc"}
        out = self.gp.build_writer_inputs("best-x", product)
        self.assertEqual(out["system"], self.gp.GENERATOR_SYSTEM_PROMPT)
        self.assertIn("Best X", out["user"])
        self.assertEqual(out["fmt"], "single_review")
        self.assertEqual(out["species"], "dog")
        self.assertNotIn("", out["user"])  # substituted

    def test_roundup_uses_products_json_runners_up_not_groq(self):
        product = {"topic": "best-mats", "title": "Best Mats", "keyword": "best mats",
                   "format": "roundup", "name": "TopMat", "category": "dogs",
                   "species": "dog", "runners_up": "MatA;MatB;MatC"}
        out = self.gp.build_writer_inputs("best-mats", product)
        self.assertIn("MatA", out["user"])
        self.assertIn("EXACTLY 3", out["user"])
        self.assertNotIn("", out["user"])

    def test_roundup_without_runners_up_uses_static_fallback_no_groq(self):
        product = {"topic": "best-mats", "title": "Best Mats", "keyword": "best mats",
                   "format": "roundup", "name": "TopMat", "category": "dogs",
                   "species": "dog"}
        out = self.gp.build_writer_inputs("best-mats", product)
        self.assertIn("well-known brands", out["user"])
        self.assertNotIn("", out["user"])

def build_writer_inputs(slug: str, product: dict) -> dict:
    """Assemble the writer's system+user prompts for one topic, internal-only.
    Roundup alternatives come from products.json `runners_up`; when absent, a
    static instruction is used instead of the (external) Groq fallback. Returns
    {system, user, title, keyword, fmt, species, affiliate_url}."""
    title    = product["title"]
    keyword  = product["keyword"]
    fmt      = product["format"]
    species  = product.get("species", "dog")
    category = product.get("category", "")

    related_url, related_anchor = find_related_published_slug(slug, category)
    user = make_prompt(title, keyword, slug, fmt, product, related_url, related_anchor)

    if "" in user:
        runners_up = product.get("runners_up", "")
        if runners_up:
            if ";" in runners_up:
                alt_count = len([a for a in runners_up.split(";") if a.strip()])
            else:
                alt_count = len([ln for ln in runners_up.splitlines() if ln.strip()])
            alt_constraint = (
                "EXACTLY " + str(alt_count) + " alternative product(s) listed below. "
                "Use ONLY these " + str(alt_count) + " product(s). "
                "Do NOT add, invent, or substitute any others.\n" + runners_up
            )
            user = user.replace("", alt_constraint)
        else:
            user = user.replace(
                "",
                "EXACTLY 3 alternatives -- use well-known brands you are confident "
                "exist. Do not fabricate products.")

    return {"system": GENERATOR_SYSTEM_PROMPT, "user": user, "title": title,
            "keyword": keyword, "fmt": fmt, "species": species,
            "affiliate_url": product.get("affiliate_url", "")}

                writer = build_writer_inputs(slug, product)
                prompt = writer["user"]
                content = call_generator(prompt, groq_key, system=writer["system"])
                log(f"  [timing] generate: {time.monotonic()-_t0:.1f}s")

class TestStageArticle(unittest.TestCase):
    def setUp(self):
        import generate_posts as gp
        self.gp = gp

    def test_stages_draft_and_pin_queue(self):
        import tempfile, json
        from pathlib import Path
        from unittest.mock import patch
        gp = self.gp
        product = {"topic": "best-x", "title": "Best X", "keyword": "best x",
                   "format": "single_review", "name": "The X", "category": "dogs",
                   "species": "dog", "affiliate_url": "https://amzn.to/abc"}
        body = "## Heading\n\n" + ("word " * 800)  # >700 words, >2000 chars
        with tempfile.TemporaryDirectory() as td:
            posts = Path(td) / "_posts"; posts.mkdir()
            pinq  = Path(td) / "_pin_queue"; pinq.mkdir()
            with patch.object(gp, "POSTS_DIR", posts), \
                 patch.object(gp, "REPO_DIR", Path(td)), \
                 patch.object(gp, "PIN_GEN_AVAILABLE", False):
                out = gp.stage_article("best-x", product, body,
                                       pin_desc="Great mat for dogs.", index=0)
            draft = posts / "DRAFT-best-x.md"
            self.assertTrue(draft.exists())
            text = draft.read_text(encoding="utf-8")
            self.assertIn('layout: post', text)
            self.assertIn('affiliate_url: "https://amzn.to/abc"', text)
            pin_json = pinq / "best-x.json"
            self.assertTrue(pin_json.exists())
            data = json.loads(pin_json.read_text(encoding="utf-8"))
            self.assertEqual(data["slug"], "best-x")
            self.assertEqual(out["draft_path"], draft)

def stage_article(slug: str, product: dict, body: str, pin_desc: str,
                  index: int = 0) -> dict:
    """Stage one passed article exactly as main() does: front-matter + pin image
    + pin-queue entry + draft-last write. `body` must be the final, review-clean
    article text (no PIN_DESC line). Returns {draft_path, pin_queue_path}."""
    title    = product["title"]
    keyword  = product["keyword"]
    species  = product.get("species", "dog")
    category = product.get("category", "")

    fm = front_matter(title, keyword, product.get("affiliate_url", ""), slug,
                      species, category, pin_desc, product.get("image", ""),
                      build_pin_image_url_for_queue(slug),
                      chewy_url=product.get("chewy_url") or "")

    content_clean = body.lstrip()
    while content_clean.startswith("---"):
        content_clean = content_clean[3:].lstrip()

    article_url = build_url(slug, utm=True)
    asin    = product.get("asin", "")
    pin_url = product.get("image", "")
    if asin:
        pin_url = f"https://images-na.ssl-images-amazon.com/images/P/{asin}.01.LZZZZZZZ.jpg"
    if PIN_GEN_AVAILABLE:
        try:
            pin_url = make_pin_for_post(title, pin_desc, pin_url, category, slug, index)
            log_pin(f"  PIN: {pin_url}")
        except Exception as pe:
            log_pin(f"  pin generation failed: {pe}", "WARN")

    pin_data = {
        "title": title, "article_url": article_url, "description": pin_desc,
        "image_url": build_pin_image_url_for_queue(slug), "species": species,
        "slug": slug, "topical_sheet": product.get("topical_sheet", ""),
    }
    draft_path = POSTS_DIR / f"DRAFT-{slugify(slug)}.md"
    pin_queue_path = REPO_DIR / "_pin_queue" / f"{slug}.json"
    persist_generated_article(draft_path, fm + "\n" + content_clean,
                              pin_queue_path, pin_data)
    log(f"  SAVED {draft_path.name} + staged pin {slug}.json")
    return {"draft_path": draft_path, "pin_queue_path": pin_queue_path}

                stage_article(slug, product, content, pin_desc, index=generated)

import json, subprocess, sys, tempfile
from pathlib import Path

PY = "./.venv/Scripts/python.exe"

def run(*args, cwd=None):
    return subprocess.run([PY, "stage1_cli.py", *args],
                          capture_output=True, text=True, cwd=cwd)

def test_gate_passes_clean_on_standard_article():
    with tempfile.TemporaryDirectory() as td:
        body = Path(td) / "body.md"; body.write_text("clean body text", encoding="utf-8")
        card = Path(td) / "card.json"
        card.write_text(json.dumps({"pass": False,
            "scores": {"human_voice": 4, "warmth": 4, "readability": 4, "accuracy": 4}}),
            encoding="utf-8")
        r = run("gate", "--body", str(body), "--scorecard", str(card))
        assert r.returncode == 0, r.stderr
        out = json.loads(r.stdout)
        assert out["passed"] is True

def test_gate_fails_on_em_dash():
    with tempfile.TemporaryDirectory() as td:
        body = Path(td) / "body.md"; body.write_text("has — dash", encoding="utf-8")
        card = Path(td) / "card.json"
        card.write_text(json.dumps({
            "scores": {"human_voice": 4, "warmth": 4, "readability": 4, "accuracy": 4}}),
            encoding="utf-8")
        r = run("gate", "--body", str(body), "--scorecard", str(card))
        out = json.loads(r.stdout)
        assert out["passed"] is False
        assert "em_dash_in_body" in out["flags"]

def test_review_prompt_contains_title_and_rubric():
    with tempfile.TemporaryDirectory() as td:
        body = Path(td) / "body.md"; body.write_text("article body", encoding="utf-8")
        r = run("review-prompt", "--body", str(body),
                "--title", "Best Dog Mats", "--keyword", "dog mats")
        assert r.returncode == 0, r.stderr
        assert "Best Dog Mats" in r.stdout
        assert "human_voice" in r.stdout

#!/usr/bin/env python3
"""Plumbing CLI for the internal Claude Stage-1 routine. Exposes the deterministic
pieces the agent needs (prompt building, the authoritative gate, staging) over a
thin command surface. No model calls happen here -- the agent supplies the write,
review, and rewrite text using Claude models."""
import argparse, json, sys
from pathlib import Path

import generate_posts as gp


def _read(path: str) -> str:
    return Path(path).read_text(encoding="utf-8")


def cmd_next_topic(args) -> int:
    products = gp.load_products()
    used = gp.build_used_slugs()
    picked = gp.select_next_topic(products, used)
    if picked is None:
        print(json.dumps({}))
        return 0
    slug, product = picked
    writer = gp.build_writer_inputs(slug, product)
    print(json.dumps({
        "slug": slug, "title": writer["title"], "keyword": writer["keyword"],
        "fmt": writer["fmt"], "species": writer["species"],
        "affiliate_url": writer["affiliate_url"],
        "system": writer["system"], "user": writer["user"],
    }))
    return 0


def cmd_review_prompt(args) -> int:
    body = _read(args.body)
    print(gp.make_review_prompt(args.title, args.keyword, body))
    return 0


def cmd_gate(args) -> int:
    body = _read(args.body)
    scorecard = json.loads(_read(args.scorecard))
    scrubbed = gp.scrub_typography(body)
    passed, flags = gp.authoritative_gate(scorecard, scrubbed)
    print(json.dumps({"passed": passed, "flags": flags, "scrubbed_body": scrubbed}))
    return 0


def cmd_rewrite_prompt(args) -> int:
    body = _read(args.body)
    instructions = _read(args.instructions)
    print(gp.make_rewrite_prompt(args.title, args.keyword, body, instructions,
                                 affiliate_url=args.affiliate_url,
                                 product_name=args.product_name))
    return 0


def cmd_stage(args) -> int:
    body = gp.scrub_typography(_read(args.body))
    products = gp.load_products()
    product = products.get(args.slug)
    if product is None:
        print(f"ERROR: unknown slug {args.slug!r}", file=sys.stderr)
        return 2
    pin_desc = gp.clean_pin_desc(args.pin_desc or f"{product['title']} - reviews and buying guide.",
                                 product.get("species", "dog"))
    try:
        gp.validate_output("review", body, args.slug,
                           affiliate_url=product.get("affiliate_url", ""))
    except gp.GenerationStageError as e:
        print(f"ERROR: content contract failed: {e}", file=sys.stderr)
        return 3
    out = gp.stage_article(args.slug, product, body, pin_desc)
    print(json.dumps({"draft_path": str(out["draft_path"]),
                      "pin_queue_path": str(out["pin_queue_path"])}))
    return 0


def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(prog="stage1_cli")
    sub = p.add_subparsers(dest="cmd", required=True)

    sub.add_parser("next-topic").set_defaults(func=cmd_next_topic)

    rp = sub.add_parser("review-prompt")
    rp.add_argument("--body", required=True); rp.add_argument("--title", required=True)
    rp.add_argument("--keyword", required=True); rp.set_defaults(func=cmd_review_prompt)

    g = sub.add_parser("gate")
    g.add_argument("--body", required=True); g.add_argument("--scorecard", required=True)
    g.set_defaults(func=cmd_gate)

    rw = sub.add_parser("rewrite-prompt")
    rw.add_argument("--body", required=True); rw.add_argument("--instructions", required=True)
    rw.add_argument("--title", default=""); rw.add_argument("--keyword", default="")
    rw.add_argument("--affiliate-url", dest="affiliate_url", default="")
    rw.add_argument("--product-name", dest="product_name", default="")
    rw.set_defaults(func=cmd_rewrite_prompt)

    st = sub.add_parser("stage")
    st.add_argument("--slug", required=True); st.add_argument("--body", required=True)
    st.add_argument("--pin-desc", dest="pin_desc", default="")
    st.set_defaults(func=cmd_stage)
    return p


def main(argv=None) -> int:
    args = build_parser().parse_args(argv)
    return args.func(args)


if __name__ == "__main__":
    sys.exit(main())