"""Stage 2 LLM dispatcher: draft a research note from an archive entry. Reads `docs/archive/.md`, assembles a prompt by stitching `scripts/librarian-prompt.md` (the system prompt) onto the archive body, and dispatches to the LLM CLI selected by the LIBRARIAN_LLM environment variable. The LLM's stdout is captured to `runs/tmp/librarian-promote--draft.md`. The dispatcher never writes into `docs/research/` directly — the operator reviews the draft and copies/edits it themselves. Backends: LIBRARIAN_LLM=claude (default; invokes `claude -p`) LIBRARIAN_LLM=codex (invokes `codex exec`) LIBRARIAN_LLM=gemini (invokes `gemini -p`) Usage: uv run python scripts/librarian_promote.py docs/archive/foo.md uv run python scripts/librarian_promote.py docs/archive/foo.md --show-prompt LIBRARIAN_LLM=codex uv run python scripts/librarian_promote.py docs/archive/foo.md """ from __future__ import annotations import argparse import os import re import subprocess import sys from datetime import datetime from pathlib import Path LLM_COMMANDS = { "claude": ["claude", "-p"], "codex": ["codex", "exec"], "gemini": ["gemini", "-p", ""], } DATE_SUFFIX = re.compile(r"-\d{4}-\d{2}-\d{2}$") LAST_VERIFIED = re.compile(r"^\*\*Last verified:\*\*[^\n]*$", re.MULTILINE) def _repo_root() -> Path: current = Path(__file__).resolve() for parent in current.parents: if (parent / "pyproject.toml").is_file(): return parent raise RuntimeError("pyproject.toml을 찾을 수 없어 repository root를 판정할 수 없습니다.") def _current_commit_sha(root: Path) -> str: """Resolve HEAD to a short SHA. Returns 'unknown' if git is unavailable.""" result = subprocess.run( ["git", "rev-parse", "--short", "HEAD"], capture_output=True, text=True, check=False, cwd=root, ) if result.returncode != 0: return "unknown" return result.stdout.strip() or "unknown" def _post_process_draft(content: str, commit_sha: str) -> str: """Normalize a draft before writing to disk. The LLM is unreliable about running `git` itself in headless mode; it tends to leave `` placeholders or literal `HEAD` strings. Rewrite the `**Last verified:**` line in place with today's date and the real short SHA. If the line is missing entirely, leave the draft untouched so the omission stays visible during review. """ today = datetime.now().strftime("%Y-%m-%d") new_line = f"**Last verified:** {today}, commit `{commit_sha}`" return LAST_VERIFIED.sub(new_line, content, count=1) def _assemble_prompt( system_prompt: str, rel_archive: Path, archive_body: str, rel_target: Path, ) -> str: return ( f"{system_prompt}\n\n" "---\n\n" "Task: Draft a `docs/research/` note from the archive entry below.\n" "Follow the rules in your system prompt above (style template, " "`Last verified:` and `Source:` headers, `file:line` citations " "verified against the current tree, ~1 page, prose over bullet " "soup).\n\n" f"**Source archive:** `{rel_archive}`\n" f"**Suggested target filename:** `{rel_target}`\n\n" "If the archive does not contain a durable conclusion (e.g. it " "is a one-off bench result with no general lesson), respond " "with a single line `SKIP: ` instead of a draft.\n\n" "Output: the markdown content of the new file only. No " "preamble, no code fences around the whole thing, no " "explanation after. Begin with the H1 header line.\n\n" "---\n\n" "Archive body:\n\n" f"{archive_body}\n" ) def main() -> int: parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument( "archive_path", help="Path to a docs/archive/*.md entry to promote.", ) parser.add_argument( "--show-prompt", action="store_true", help="Print the assembled prompt to stdout and exit; do not call the LLM.", ) parser.add_argument( "--accept", action="store_true", help=( "After the draft lands in runs/tmp/, also copy it verbatim to the " "suggested docs/research/ target. Explicit per-invocation opt-in: " "still propose-only by design (you are choosing acceptance " "knowingly), not auto-apply." ), ) args = parser.parse_args() root = _repo_root() archive = (root / args.archive_path).resolve() if not archive.is_file(): print(f"Archive not found: {args.archive_path}", file=sys.stderr) return 1 try: rel_archive = archive.relative_to(root) except ValueError: print(f"Archive must live under repo root: {archive}", file=sys.stderr) return 1 if not str(rel_archive).startswith("docs/archive/"): print( f"Refusing: archive must live under docs/archive/: {rel_archive}", file=sys.stderr, ) return 1 stem = DATE_SUFFIX.sub("", archive.stem) target = root / "docs" / "research" / f"{stem}.md" rel_target = target.relative_to(root) if target.exists(): print( f"Refusing: research counterpart already exists: {rel_target}", file=sys.stderr, ) return 1 system_prompt = (root / "scripts" / "librarian-prompt.md").read_text(encoding="utf-8") archive_body = archive.read_text(encoding="utf-8") prompt = _assemble_prompt(system_prompt, rel_archive, archive_body, rel_target) if args.show_prompt: sys.stdout.write(prompt) return 0 backend = os.environ.get("LIBRARIAN_LLM", "claude").lower() cmd = LLM_COMMANDS.get(backend) if cmd is None: print( f"Unknown LIBRARIAN_LLM={backend}; supported: {sorted(LLM_COMMANDS)}", file=sys.stderr, ) return 1 timestamp = datetime.now().strftime("%Y-%m-%d_%H%M%S") out_dir = root / "runs" / "tmp" out_dir.mkdir(parents=True, exist_ok=True) prompt_path = out_dir / f"librarian-promote-{timestamp}-prompt.md" draft_path = out_dir / f"librarian-promote-{timestamp}-draft.md" prompt_path.write_text(prompt, encoding="utf-8") print( f"Dispatching to {backend} (prompt saved to {prompt_path.relative_to(root)})", file=sys.stderr, ) result = subprocess.run( cmd, input=prompt, capture_output=True, text=True, check=False, ) if result.returncode != 0: print(f"LLM call failed (exit {result.returncode}):", file=sys.stderr) print(result.stderr, file=sys.stderr) return result.returncode commit_sha = _current_commit_sha(root) draft_text = _post_process_draft(result.stdout, commit_sha) draft_path.write_text(draft_text, encoding="utf-8") print(f"Draft written to: {draft_path.relative_to(root)}") if args.accept: if target.exists(): print( f"Refusing --accept: {rel_target} appeared while the LLM ran.", file=sys.stderr, ) return 1 target.write_text(draft_text, encoding="utf-8") print(f"Copied to: {rel_target}") print("Next: review the diff and `git add` + commit if you're satisfied.") else: print(f"Suggested target on accept: {rel_target}") print() print("Next: review the draft. To accept verbatim:") print(f" cp {draft_path.relative_to(root)} {rel_target}") print("Or rerun with --accept to copy in the same step.") return 0 if __name__ == "__main__": raise SystemExit(main())