134 lines
4.6 KiB
Python
134 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
AI_Moderator agent.
|
|
|
|
Upgraded to use a persona-driven AI prompt to contribute meaningful comments and
|
|
votes to discussions, rather than just simple reminders.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# --- Dynamic SDK Loader ---
|
|
|
|
def load_agent_context(repo_root: Path):
|
|
vendor_root = repo_root / ".cascadingdev"
|
|
if (vendor_root / "cascadingdev" / "agent").exists():
|
|
sys.path.insert(0, str(vendor_root))
|
|
from cascadingdev.agent.sdk import AgentContext
|
|
return AgentContext
|
|
project_root = Path(__file__).resolve().parents[1]
|
|
src_path = project_root / "src"
|
|
if src_path.exists():
|
|
sys.path.insert(0, str(src_path))
|
|
from cascadingdev.agent.sdk import AgentContext
|
|
return AgentContext
|
|
from cascadingdev.agent.sdk import AgentContext
|
|
return AgentContext
|
|
|
|
def load_provider_client(repo_root: Path):
|
|
vendor_root = repo_root / ".cascadingdev"
|
|
if (vendor_root / "cascadingdev" / "agent").exists():
|
|
sys.path.insert(0, str(vendor_root))
|
|
from cascadingdev.agent.providers import ProviderClient
|
|
return ProviderClient
|
|
project_root = Path(__file__).resolve().parents[1]
|
|
src_path = project_root / "src"
|
|
if src_path.exists():
|
|
sys.path.insert(0, str(src_path))
|
|
from cascadingdev.agent.providers import ProviderClient
|
|
return ProviderClient
|
|
from cascadingdev.agent.providers import ProviderClient
|
|
return ProviderClient
|
|
|
|
# --- Agent Configuration ---
|
|
|
|
PRIMER = """
|
|
You are AI_Moderator, the project steward for CascadingDev discussions.
|
|
|
|
Responsibilities:
|
|
- When a new discussion starts, post an introductory note outlining objectives for the stage.
|
|
- Monitor progression, call out missing votes or unclear decisions, and suggest next steps.
|
|
- Vote READY only when the goals are satisfied; otherwise vote CHANGES and explain what is missing.
|
|
|
|
Instructions:
|
|
- Read the current discussion content carefully.
|
|
- Respond with a JSON object containing keys: "comment" (Markdown body only, no Name/VOTE lines), "vote" (READY/CHANGES/REJECT).
|
|
- If no action is required, set "comment" to an empty string but still include "vote".
|
|
- Keep the tone collaborative and concise.
|
|
"""
|
|
|
|
def build_prompt(discussion_text: str) -> str:
|
|
return f"{PRIMER}\n\nCurrent discussion:\n\n{discussion_text}\n\nProvide your JSON response now."
|
|
|
|
def scrub_comment(text: str) -> str:
|
|
cleaned: list[str] = []
|
|
for line in text.splitlines():
|
|
stripped = line.strip()
|
|
upper = stripped.upper()
|
|
if upper.startswith("NAME:"):
|
|
continue
|
|
if upper.startswith("VOTE:"):
|
|
continue
|
|
cleaned.append(line)
|
|
return "\n".join(cleaned).strip()
|
|
|
|
def append_moderator_comment(context, comment: str, vote: str) -> None:
|
|
lines = ["---", "", "Name: AI_Moderator"]
|
|
comment = scrub_comment(comment)
|
|
if comment:
|
|
lines.append(comment)
|
|
lines.append(f"VOTE: {vote}")
|
|
lines.append("")
|
|
context.append_block("\n".join(lines))
|
|
|
|
def ensure_staged(repo_root: Path, relative_path: Path) -> None:
|
|
subprocess.run(["git", "add", relative_path.as_posix()], cwd=repo_root, check=False, capture_output=True)
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="AI Moderator agent")
|
|
parser.add_argument("--repo-root", required=True, help="Repository root path")
|
|
parser.add_argument("--path", required=True, help="Relative path to discussion file")
|
|
args = parser.parse_args()
|
|
|
|
repo_root = Path(args.repo_root).resolve()
|
|
discussion_rel = Path(args.path)
|
|
|
|
AgentContext = load_agent_context(repo_root)
|
|
ProviderClient = load_provider_client(repo_root)
|
|
|
|
context = AgentContext(repo_root, discussion_rel)
|
|
|
|
text = context.read_text()
|
|
if not text.strip():
|
|
return 0
|
|
|
|
provider = ProviderClient(repo_root)
|
|
try:
|
|
response = provider.structured(build_prompt(text), model_hint="quality")
|
|
except Exception:
|
|
response = {
|
|
"comment": "Gentle reminder: please keep the discussion moving and ensure votes reflect actual consensus.",
|
|
"vote": "CHANGES",
|
|
}
|
|
|
|
comment = str(response.get("comment", "")).strip()
|
|
vote = str(response.get("vote", "READY")).strip().upper()
|
|
if vote not in {"READY", "CHANGES", "REJECT"}:
|
|
vote = "CHANGES"
|
|
|
|
# Do nothing if the AI returns an empty comment and a non-binding vote
|
|
if not comment and vote == "READY":
|
|
return 0
|
|
|
|
append_moderator_comment(context, comment, vote)
|
|
ensure_staged(repo_root, discussion_rel)
|
|
return 0
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|