#!/usr/bin/env python3 """Turn a scan_x_profile.py corpus into durable author-voice exemplars. WHERE THIS FITS: scan_x_profile.py (run at onboarding right after connect_x, or later via build_persona.py gather) returns the user's real posts + replies ranked by engagement (`top_posts` / `top_replies`, each with likes/retweets/ replies, the parent tweet replied to, and thread continuations). This script is the single deterministic writer that persists them where the drafters already look: 1. config.json -> project.voice.examples (list of strings). Every drafting prompt already says "mirror voice.examples / voice.examples_good when present": the twitter cycle (both promotion and personal_brand lanes, via ALL_PROJECTS_JSON), post_reddit.py, engage_reddit.py, post_github.py, and run_moltbook_cycle.py. Writing this ONE field feeds every lane on every platform with zero prompt changes. 2. persona_corpus.txt (persona project only) -> a marked, regenerable exemplar section with the top posts (incl. thread continuations) and top replies WITH their parent context and stats. The personal_brand lane inlines this file into the draft prompt. Selection: prefers replies with real engagement and enough text to show voice (skips throwaway one-liners below --min-chars unless nothing else exists). Stats are framed RELATIVELY ("top of the N scanned replies") because absolute numbers on small accounts mean nothing to the model. Safety: never clobbers hand-written voice.examples. It only replaces examples it wrote itself (voice.examples_scanned_at present) or fills an empty field; use --force to override. Config writes are atomic (tmp + replace), matching build_persona.py. Usage: # After a scan (writes voice.examples; also corpus section if project is the persona): python3 scripts/scan_x_profile.py > /tmp/scan.json python3 scripts/voice_exemplars.py apply --scan /tmp/scan.json --project MyProject # Preview without writing: python3 scripts/voice_exemplars.py apply --scan /tmp/scan.json --project MyProject --dry-run # Default project = the persona project (persona:true) when --project omitted. Accepts either raw scan_x_profile.py output or a build_persona.py gather blob (auto-detects sources.x). """ from __future__ import annotations import argparse import json import os import shutil import subprocess import sys import time from datetime import datetime, timezone from pathlib import Path HERE = Path(__file__).resolve().parent sys.path.insert(0, str(HERE)) import s4l_mode # noqa: E402 from scan_x_profile import SCANNER_VERSION # noqa: E402 EXEMPLARS_START = "=== S4L SCANNED EXEMPLARS START (auto-generated by voice_exemplars.py; regenerated on rescan, do not hand-edit) ===" EXEMPLARS_END = "=== S4L SCANNED EXEMPLARS END ===" # Periodic staleness refresh (2026-07-16): even a scan done with the current # scanner can go stale as the account posts/replies more over time. Re-scan # any persona project whose last scan is older than this, on top of the # version-triggered rescan below. RESCAN_MAX_AGE_DAYS = 14 def _needs_rescan(proj: dict) -> bool: """True if this persona project should get a fresh profile scan: never scanned, hand-written examples (exempt from auto-touch, but also never "need" a rescan since we'd never apply one), the last scan predates the current SCANNER_VERSION (a scraping/ranking fix shipped since), or the last scan is older than RESCAN_MAX_AGE_DAYS. An unparseable timestamp is treated as stale rather than silently never-rescanned.""" if _is_hand_written(proj): return False voice = proj.get("voice") if not isinstance(voice, dict) or not voice.get("examples_scanned_at"): return True if int(voice.get("examples_scanner_version") or 0) < SCANNER_VERSION: return True try: scanned_at = datetime.fromisoformat(voice["examples_scanned_at"]) if scanned_at.tzinfo is None: scanned_at = scanned_at.replace(tzinfo=timezone.utc) age_days = (datetime.now(timezone.utc) - scanned_at).days return age_days >= RESCAN_MAX_AGE_DAYS except Exception: return True def _score(item: dict) -> int: """Mirror scan_x_profile.rank_top's weighting for scans that predate it.""" if "engagement_score" in item: return int(item["engagement_score"] or 0) return (int(item.get("likes") or 0) * 3 + int(item.get("retweets") or 0) * 5 + int(item.get("replies") or 0) * 2) def _load_scan(path: str) -> dict: data = json.loads(Path(path).expanduser().read_text()) # build_persona.py gather embeds the raw scan under sources.x if "sources" in data and isinstance(data["sources"], dict): x = data["sources"].get("x") or {} if x.get("ok"): return x raise SystemExit(f"gather blob's x source is not ok: {x.get('error')}") return data def _pick(items: list, top: int, min_chars: int) -> list: """Engagement-ranked, but a reply must have enough text to show voice. Short high-engagement one-liners get skipped while longer candidates exist; if the account only has one-liners, fall back rather than return nothing.""" ranked = sorted(items or [], key=_score, reverse=True) long_enough = [i for i in ranked if len((i.get("text") or "").strip()) >= min_chars] picked = long_enough[:top] if not picked: picked = ranked[:top] return picked def _reply_pool(scan: dict) -> list: """Select from the FULL scanned reply list, not the scanner's pre-truncated top_replies (length-filtering a 5-item list can leave one survivor when the account's best-engaging replies are one-liners). top_replies is the fallback for older scans / gather blobs that dropped the full list.""" return scan.get("comments") or scan.get("top_replies") or [] def _post_pool(scan: dict) -> list: """Full post list, but prefer the enriched top_posts entry where one exists (it carries the thread continuation and untruncated permalink text).""" top = scan.get("top_posts") or [] rest = scan.get("posts") or [] by_key = {(p.get("id") or p.get("url") or p.get("text", "")[:80]): p for p in rest} for p in top: by_key[(p.get("id") or p.get("url") or p.get("text", "")[:80])] = p return list(by_key.values()) def _stats_note(item: dict, total: int, kind: str) -> str: bits = [] for key, label in (("likes", "likes"), ("retweets", "reposts"), ("replies", "replies")): v = int(item.get(key) or 0) if v: bits.append(f"{v} {label}") eng = ", ".join(bits) if bits else "no engagement yet" return f"{eng}; among their best of {total} scanned {kind}" def format_examples(scan: dict, top: int, min_chars: int) -> list[str]: """voice.examples strings: verbatim reply first (the voice signal), then a parenthesized context annotation (real stats + what it replied to). Plain strings because every consumer joins/inlines them as-is.""" replies = _reply_pool(scan) total = int((scan.get("counts") or {}).get("comments") or len(replies)) out = [] for r in _pick(replies, top, min_chars): text = (r.get("text") or "").strip() if not text: continue note = _stats_note(r, total, "replies") parent = r.get("parent") or {} if parent.get("text"): ptxt = " ".join(parent["text"].split())[:140] note += f'; replying to {parent.get("author") or "someone"}: "{ptxt}"' out.append(f'"{text}" (their real reply, {note})') return out def format_corpus_section(scan: dict, top: int, min_chars: int) -> str: """The persona_corpus.txt block: richer than voice.examples (adds top posts and full thread continuations). Regenerated wholesale between the markers.""" handle = scan.get("handle") or "?" counts = scan.get("counts") or {} lines = [EXEMPLARS_START, f"@{handle}'s own top-performing posts and replies, ranked by real " f"engagement across {counts.get('posts', '?')} posts and " f"{counts.get('comments', '?')} replies scanned. Verbatim. This is " "the voice and the bar: ground drafts in this register.", ""] posts = _pick(_post_pool(scan), top, min_chars) if posts: lines.append("TOP POSTS:") for p in posts: lines.append(f"- ({_stats_note(p, int(counts.get('posts') or len(posts)), 'posts')}) {p.get('text', '').strip()}") for cont in (p.get("thread") or [])[1:]: lines.append(f" [thread continuation] {cont.strip()}") lines.append("") replies = _pick(_reply_pool(scan), top, min_chars) if replies: lines.append("TOP REPLIES (with what they replied to):") for r in replies: parent = r.get("parent") or {} if parent.get("text"): ptxt = " ".join(parent["text"].split())[:200] lines.append(f'- {parent.get("author") or "someone"} wrote: "{ptxt}"') lines.append(f" their reply ({_stats_note(r, int(counts.get('comments') or len(replies)), 'replies')}): {r.get('text', '').strip()}") else: lines.append(f"- their reply ({_stats_note(r, int(counts.get('comments') or len(replies)), 'replies')}): {r.get('text', '').strip()}") lines.append(EXEMPLARS_END) return "\n".join(lines) def _upsert_corpus_section(corpus_path: Path, section: str) -> str: """Replace the marked section in persona_corpus.txt, or prepend it (top of file for salience) when absent. Returns what happened.""" if corpus_path.exists(): body = corpus_path.read_text() if EXEMPLARS_START in body and EXEMPLARS_END in body: pre = body.split(EXEMPLARS_START)[0] post = body.split(EXEMPLARS_END, 1)[1] corpus_path.write_text((pre + section + post).strip() + "\n") return "replaced" corpus_path.write_text(section + "\n\n" + body.strip() + "\n") return "prepended" corpus_path.write_text(section + "\n") return "created" def _is_hand_written(proj: dict) -> bool: """examples present but never stamped by us = a human typed them.""" voice = proj.get("voice") return (isinstance(voice, dict) and bool(voice.get("examples")) and not voice.get("examples_scanned_at")) def _set_examples(proj: dict, scan: dict, top: int, min_chars: int, force: bool) -> "tuple[bool, str, list]": """Write voice.examples onto proj IN PLACE. Returns (ok, reason, examples).""" examples = format_examples(scan, top, min_chars) if not examples: return False, "no_usable_replies", [] if _is_hand_written(proj) and not force: return False, "hand_written_examples", [] voice = proj.get("voice") if not isinstance(voice, dict): voice = {"tone": voice} if voice else {} proj["voice"] = voice voice["examples"] = examples voice["examples_scanned_at"] = datetime.now(timezone.utc).isoformat(timespec="seconds") voice["examples_scanner_version"] = scan.get("scanner_version") or SCANNER_VERSION return True, "ok", examples def _write_config(cfg_path: Path, cfg: dict) -> None: tmp = cfg_path.with_suffix(".json.tmp") tmp.write_text(json.dumps(cfg, ensure_ascii=False, indent=2) + "\n") tmp.replace(cfg_path) def cmd_apply(args) -> int: cfg_path = Path(args.config).expanduser() if args.config else s4l_mode.config_path() scan_src = args.scan or str(cfg_path.parent / "last_profile_scan.json") if not Path(scan_src).expanduser().exists(): print(json.dumps({"ok": False, "error": f"no scan at {scan_src}; run " "scan_x_profile.py first (it writes last_profile_scan.json) " "or pass --scan"})) return 1 scan = _load_scan(scan_src) if not scan.get("ok"): print(json.dumps({"ok": False, "error": f"scan not ok: {scan.get('error')}"})) return 1 cfg = json.loads(cfg_path.read_text()) projects = cfg.get("projects", []) name = args.project or s4l_mode.persona_name() proj = next((p for p in projects if p.get("name") == name), None) if proj is None: print(json.dumps({"ok": False, "error": f"project {name!r} not in {cfg_path} " "(pass --project, or configure a persona project)"})) return 2 ok, reason, examples = _set_examples(proj, scan, args.top, args.min_chars, args.force) if not ok and reason == "no_usable_replies": print(json.dumps({"ok": False, "error": "scan has no usable replies to build examples from"})) return 1 if not ok: print(json.dumps({"ok": False, "error": f"{name}.voice.examples already has hand-written entries; " "rerun with --force to replace them", "existing": (proj.get("voice") or {}).get("examples")})) return 3 is_persona = proj.get("persona") is True corpus_path = cfg_path.parent / "persona_corpus.txt" section = format_corpus_section(scan, args.top, args.min_chars) if is_persona else None if args.dry_run: print(json.dumps({"ok": True, "dry_run": True, "project": name, "voice_examples": examples, "corpus_section": section}, ensure_ascii=False, indent=2)) return 0 _write_config(cfg_path, cfg) corpus_action = None if section: corpus_action = _upsert_corpus_section(corpus_path, section) print(json.dumps({"ok": True, "project": name, "voice_examples_written": len(examples), "corpus_sidecar": str(corpus_path) if corpus_action else None, "corpus_action": corpus_action}, ensure_ascii=False)) return 0 # --------------------------------------------------------------------------- # # backfill: one-shot catch-up for installs onboarded BEFORE this feature. # --------------------------------------------------------------------------- # _BROWSER_LOCK = Path("/tmp/social-autoposter-twitter-browser.lock") def _try_browser_lock() -> "Path | None": """One mkdir attempt on the pipelines' twitter-browser lock (same protocol as skill/lock.sh, without its queue).""" try: _BROWSER_LOCK.mkdir() except OSError: return None try: (_BROWSER_LOCK / "pid").write_text(f"{os.getpid()}\n") (_BROWSER_LOCK / "expires_at").write_text(f"{int(time.time()) + 600}\n") except Exception: pass return _BROWSER_LOCK def _wait_browser_lock(max_wait_s: float, poll_s: float = 20.0) -> "Path | None": """Wait for the twitter-browser lock however long it takes (up to max_wait_s). Holds no other lock while waiting, so this cannot deadlock; it just queues politely behind cycles / DM runs. Reclaims a stale lock (holder pid dead + lease expired), mirroring skill/lock.sh.""" deadline = time.time() + max_wait_s while True: lock = _try_browser_lock() if lock is not None: return lock try: pid = int((_BROWSER_LOCK / "pid").read_text().strip()) expires = int((_BROWSER_LOCK / "expires_at").read_text().strip()) pid_alive = True try: os.kill(pid, 0) except OSError: pid_alive = False if not pid_alive and time.time() > expires: shutil.rmtree(_BROWSER_LOCK, ignore_errors=True) continue # retry mkdir immediately except Exception: pass # lock mid-transition; just poll again if time.time() >= deadline: return None time.sleep(poll_s) def _release_browser_lock() -> None: try: pid = (_BROWSER_LOCK / "pid").read_text().strip() if pid == str(os.getpid()): shutil.rmtree(_BROWSER_LOCK) except Exception: pass def cmd_backfill(args) -> int: """Called best-effort on MCP boot. Runs a fresh profile scan and (re)applies exemplars for any persona project that _needs_rescan() flags: never scanned (predates the exemplar feature), scanned by an older SCANNER_VERSION (a scraping/ranking fix shipped since -- e.g. the 2026-07-15 stall-detection fix silently undercounted replies on accounts where S4L posts heavily, so those pre-fix scans need redoing even though they "succeeded"), or scanned more than RESCAN_MAX_AGE_DAYS ago (2026-07-16: periodic staleness refresh, since the account keeps posting/replying after the last scan). Scoped to the PERSONA project only: every install provisions one at onboarding, and it is where the author's own voice lives; product projects pick exemplars up on their next project_config save instead. The corpus write is additive (the marked section is the only thing replaced; dictation and hand-added material stay). No cooldown beyond the staleness window by design (user rule 2026-07-10 extended 2026-07-16): every boot retries until the scan succeeds; success stamps examples_scanned_at + examples_scanner_version, which makes all later boots a cheap no-op UNTIL either goes stale again. Boots are user-triggered (Desktop launch), so the worst case is one scan attempt per launch, and concurrent waiters from overlapping boots dedupe via the post-lock eligibility re-check.""" cfg_path = Path(args.config).expanduser() if args.config else s4l_mode.config_path() if not cfg_path.exists(): print(json.dumps({"ok": True, "did": "nothing", "reason": "no_config"})) return 0 cfg = json.loads(cfg_path.read_text()) personas = [p for p in cfg.get("projects", []) if p.get("persona") is True] eligible = [p for p in personas if _needs_rescan(p)] if not eligible: print(json.dumps({"ok": True, "did": "nothing", "reason": "already_done_or_hand_written"})) return 0 if not args.no_scan: lock = _wait_browser_lock(args.max_wait_minutes * 60) if lock is None: # Waited the whole window and never got the browser; the next # boot simply tries again. print(json.dumps({"ok": True, "did": "nothing", "reason": "browser_busy_timeout"})) return 0 # The wait can be hours; another boot's backfill may have finished # meanwhile. Re-read config and re-check before burning a scan. cfg = json.loads(cfg_path.read_text()) personas = [p for p in cfg.get("projects", []) if p.get("persona") is True] eligible = [p for p in personas if _needs_rescan(p)] if not eligible: _release_browser_lock() print(json.dumps({"ok": True, "did": "nothing", "reason": "done_while_waiting"})) return 0 try: py = os.environ.get("S4L_PYTHON") or sys.executable or "python3" # generous: the default depth (60 posts / 150 replies) can scroll # for several minutes on prolific accounts r = subprocess.run([py, str(HERE / "scan_x_profile.py")], capture_output=True, text=True, timeout=1200) except Exception as e: print(json.dumps({"ok": False, "did": "nothing", "reason": f"scan_failed: {e}"})) return 0 finally: _release_browser_lock() last_line = (r.stdout or "").strip().splitlines()[-1:] or ["{}"] try: scan_res = json.loads(last_line[0]) except Exception: scan_res = {} if not scan_res.get("ok"): print(json.dumps({"ok": False, "did": "nothing", "reason": f"scan_not_ok: {scan_res.get('state') or 'no_json'}"})) return 0 scan_path = cfg_path.parent / "last_profile_scan.json" if not scan_path.exists(): print(json.dumps({"ok": False, "did": "nothing", "reason": "no_scan_sidecar"})) return 0 scan = _load_scan(str(scan_path)) applied = [] for proj in eligible: ok, reason, examples = _set_examples(proj, scan, args.top, args.min_chars, False) if ok: applied.append(proj["name"]) if proj.get("persona") is True: section = format_corpus_section(scan, args.top, args.min_chars) _upsert_corpus_section(cfg_path.parent / "persona_corpus.txt", section) if applied: _write_config(cfg_path, cfg) print(json.dumps({"ok": True, "did": "backfilled" if applied else "nothing", "projects": applied}, ensure_ascii=False)) return 0 def main(argv) -> int: ap = argparse.ArgumentParser(description="Persist scanned top posts/replies as author-voice exemplars") sub = ap.add_subparsers(dest="cmd", required=True) a = sub.add_parser("apply", help="write voice.examples (+ persona corpus section)") a.add_argument("--scan", default=None, help="scan_x_profile.py JSON or build_persona gather blob " "(default: /last_profile_scan.json, written by the scanner)") a.add_argument("--project", default=None, help="config.json project name (default: the persona project)") a.add_argument("--config", default=None, help="config.json path override (testing)") a.add_argument("--top", type=int, default=5, help="max exemplars per surface") a.add_argument("--min-chars", type=int, default=40, help="skip replies shorter than this when longer ones exist") a.add_argument("--force", action="store_true", help="replace hand-written voice.examples") a.add_argument("--dry-run", action="store_true", help="print what would be written") a.set_defaults(func=cmd_apply) b = sub.add_parser("backfill", help="one-shot catch-up on boot: scan + apply for a persona " "that predates the exemplar feature") b.add_argument("--config", default=None, help="config.json path override (testing)") b.add_argument("--top", type=int, default=5) b.add_argument("--min-chars", type=int, default=40) b.add_argument("--max-wait-minutes", type=float, default=12 * 60, help="how long to wait for the twitter-browser lock before " "giving up until the next boot") b.add_argument("--no-scan", action="store_true", help="skip the live scan and use the existing last_profile_scan.json") b.set_defaults(func=cmd_backfill) args = ap.parse_args(argv) return args.func(args) if __name__ == "__main__": raise SystemExit(main(sys.argv[1:]))