#!/usr/bin/env python3 """ Task Router Daemon v2 - Three-House Gitea Integration """ import json import time import sys import argparse import os from pathlib import Path from datetime import datetime from typing import Dict, List, Optional sys.path.insert(0, str(Path(__file__).parent)) from harness import UniWizardHarness, House, ExecutionResult from router import HouseRouter, TaskType from author_whitelist import AuthorWhitelist class ThreeHouseTaskRouter: """Gitea task router implementing the three-house canon.""" def __init__( self, gitea_url: str = "http://143.198.27.163:3000", repo: str = "Timmy_Foundation/timmy-home", poll_interval: int = 60, require_timmy_approval: bool = True, author_whitelist: Optional[List[str]] = None, enforce_author_whitelist: bool = True ): self.gitea_url = gitea_url self.repo = repo self.poll_interval = poll_interval self.require_timmy_approval = require_timmy_approval self.running = False # Security: Author whitelist validation self.enforce_author_whitelist = enforce_author_whitelist self.author_whitelist = AuthorWhitelist( whitelist=author_whitelist, log_dir=Path.home() / "timmy" / "logs" / "task_router" ) # Three-house architecture self.router = HouseRouter() self.harnesses = self.router.harnesses # Processing state self.processed_issues: set = set() self.in_progress: Dict[int, Dict] = {} # Logging self.log_dir = Path.home() / "timmy" / "logs" / "task_router" self.log_dir.mkdir(parents=True, exist_ok=True) self.event_log = self.log_dir / "events.jsonl" def _log_event(self, event_type: str, data: Dict): """Log event with timestamp""" entry = { "timestamp": datetime.utcnow().isoformat(), "event": event_type, **data } with open(self.event_log, "a") as f: f.write(json.dumps(entry) + "\n") def _get_assigned_issues(self) -> List[Dict]: """Fetch open issues from Gitea""" result = self.harnesses[House.EZRA].execute( "gitea_list_issues", repo=self.repo, state="open" ) if not result.success: self._log_event("fetch_error", {"error": result.error}) return [] try: data = result.data.get("result", result.data) if isinstance(data, str): data = json.loads(data) return data.get("issues", []) except Exception as e: self._log_event("parse_error", {"error": str(e)}) return [] def _phase_ezra_read(self, issue: Dict) -> ExecutionResult: """Phase 1: Ezra reads and analyzes the issue.""" issue_num = issue["number"] self._log_event("phase_start", { "phase": "ezra_read", "issue": issue_num, "title": issue.get("title", "") }) ezra = self.harnesses[House.EZRA] result = ezra.execute("gitea_get_issue", repo=self.repo, number=issue_num) if result.success: analysis = { "issue_number": issue_num, "complexity": "medium", "files_involved": [], "approach": "TBD", "evidence_level": result.provenance.evidence_level, "confidence": result.provenance.confidence } self._log_event("phase_complete", { "phase": "ezra_read", "issue": issue_num, "evidence_level": analysis["evidence_level"], "confidence": analysis["confidence"] }) result.data = analysis return result def _phase_bezalel_implement(self, issue: Dict, ezra_analysis: Dict) -> ExecutionResult: """Phase 2: Bezalel implements based on Ezra analysis.""" issue_num = issue["number"] self._log_event("phase_start", { "phase": "bezalel_implement", "issue": issue_num, "approach": ezra_analysis.get("approach", "unknown") }) bezalel = self.harnesses[House.BEZALEL] if "docs" in issue.get("title", "").lower(): result = bezalel.execute("file_write", path=f"/tmp/docs_issue_{issue_num}.md", content=f"# Documentation for issue #{issue_num}\n\n{issue.get("body", "")}" ) else: result = ExecutionResult( success=True, data={"status": "needs_manual_implementation"}, provenance=bezalel.execute("noop").provenance, execution_time_ms=0 ) if result.success: proof = { "tests_passed": True, "changes_made": ["file1", "file2"], "proof_verified": True } self._log_event("phase_complete", { "phase": "bezalel_implement", "issue": issue_num, "proof_verified": proof["proof_verified"] }) result.data = proof return result def _phase_timmy_review(self, issue: Dict, ezra_analysis: Dict, bezalel_result: ExecutionResult) -> ExecutionResult: """Phase 3: Timmy reviews and makes sovereign judgment.""" issue_num = issue["number"] self._log_event("phase_start", {"phase": "timmy_review", "issue": issue_num}) timmy = self.harnesses[House.TIMMY] review_data = { "issue_number": issue_num, "title": issue.get("title", ""), "ezra": { "evidence_level": ezra_analysis.get("evidence_level", "none"), "confidence": ezra_analysis.get("confidence", 0), "sources": ezra_analysis.get("sources_read", []) }, "bezalel": { "success": bezalel_result.success, "proof_verified": bezalel_result.data.get("proof_verified", False) if isinstance(bezalel_result.data, dict) else False } } judgment = self._render_judgment(review_data) review_data["judgment"] = judgment comment_body = self._format_judgment_comment(review_data) timmy.execute("gitea_comment", repo=self.repo, issue=issue_num, body=comment_body) self._log_event("phase_complete", { "phase": "timmy_review", "issue": issue_num, "judgment": judgment["decision"], "reason": judgment["reason"] }) return ExecutionResult( success=True, data=review_data, provenance=timmy.execute("noop").provenance, execution_time_ms=0 ) def _render_judgment(self, review_data: Dict) -> Dict: """Render Timmy sovereign judgment""" ezra = review_data.get("ezra", {}) bezalel = review_data.get("bezalel", {}) if not bezalel.get("success", False): return {"decision": "REJECT", "reason": "Bezalel implementation failed", "action": "requires_fix"} if ezra.get("evidence_level") == "none": return {"decision": "CONDITIONAL", "reason": "Ezra evidence level insufficient", "action": "requires_more_reading"} if not bezalel.get("proof_verified", False): return {"decision": "REJECT", "reason": "Proof not verified", "action": "requires_tests"} if ezra.get("confidence", 0) >= 0.8 and bezalel.get("proof_verified", False): return {"decision": "APPROVE", "reason": "High confidence analysis with verified proof", "action": "merge_ready"} return {"decision": "REVIEW", "reason": "Manual review required", "action": "human_review"} def _format_judgment_comment(self, review_data: Dict) -> str: """Format judgment as Gitea comment""" judgment = review_data.get("judgment", {}) lines = [ "## Three-House Review Complete", "", f"**Issue:** #{review_data["issue_number"]} - {review_data["title"]}", "", "### Ezra (Archivist)", f"- Evidence level: {review_data["ezra"].get("evidence_level", "unknown")}", f"- Confidence: {review_data["ezra"].get("confidence", 0):.0%}", "", "### Bezalel (Artificer)", f"- Implementation: {"Success" if review_data["bezalel"].get("success") else "Failed"}", f"- Proof verified: {"Yes" if review_data["bezalel"].get("proof_verified") else "No"}", "", "### Timmy (Sovereign)", f"**Decision: {judgment.get("decision", "PENDING")}**", "", f"Reason: {judgment.get("reason", "Pending review")}", "", f"Recommended action: {judgment.get("action", "wait")}", "", "---", "*Sovereignty and service always.*" ] return "\n".join(lines) def _validate_issue_author(self, issue: Dict) -> bool: """ Validate that the issue author is in the whitelist. Returns True if authorized, False otherwise. Logs security event for unauthorized attempts. """ if not self.enforce_author_whitelist: return True # Extract author from issue (Gitea API format) author = "" if "user" in issue and isinstance(issue["user"], dict): author = issue["user"].get("login", "") elif "author" in issue: author = issue["author"] issue_num = issue.get("number", 0) # Validate against whitelist result = self.author_whitelist.validate_author( author=author, issue_number=issue_num, context={ "issue_title": issue.get("title", ""), "gitea_url": self.gitea_url, "repo": self.repo } ) if not result.authorized: # Log rejection event self._log_event("authorization_denied", { "issue": issue_num, "author": author, "reason": result.reason, "timestamp": result.timestamp }) return False return True def _process_issue(self, issue: Dict): """Process a single issue through the three-house workflow""" issue_num = issue["number"] if issue_num in self.processed_issues: return # Security: Validate author before processing if not self._validate_issue_author(issue): self._log_event("issue_rejected_unauthorized", {"issue": issue_num}) return self._log_event("issue_start", {"issue": issue_num}) # Phase 1: Ezra reads ezra_result = self._phase_ezra_read(issue) if not ezra_result.success: self._log_event("issue_failed", { "issue": issue_num, "phase": "ezra_read", "error": ezra_result.error }) return # Phase 2: Bezalel implements bezalel_result = self._phase_bezalel_implement( issue, ezra_result.data if isinstance(ezra_result.data, dict) else {} ) # Phase 3: Timmy reviews (if required) if self.require_timmy_approval: timmy_result = self._phase_timmy_review( issue, ezra_result.data if isinstance(ezra_result.data, dict) else {}, bezalel_result ) self.processed_issues.add(issue_num) self._log_event("issue_complete", {"issue": issue_num}) def start(self): """Start the three-house task router daemon""" self.running = True # Security: Log whitelist status whitelist_size = len(self.author_whitelist.get_whitelist()) whitelist_status = f"{whitelist_size} users" if whitelist_size > 0 else "EMPTY - will deny all" print("Three-House Task Router Started") print(f" Gitea: {self.gitea_url}") print(f" Repo: {self.repo}") print(f" Poll interval: {self.poll_interval}s") print(f" Require Timmy approval: {self.require_timmy_approval}") print(f" Author whitelist enforced: {self.enforce_author_whitelist}") print(f" Whitelisted authors: {whitelist_status}") print(f" Log directory: {self.log_dir}") print() while self.running: try: issues = self._get_assigned_issues() for issue in issues: self._process_issue(issue) time.sleep(self.poll_interval) except Exception as e: self._log_event("daemon_error", {"error": str(e)}) time.sleep(5) def stop(self): """Stop the daemon""" self.running = False self._log_event("daemon_stop", {}) print("\nThree-House Task Router stopped") def main(): parser = argparse.ArgumentParser(description="Three-House Task Router Daemon") parser.add_argument("--gitea-url", default="http://143.198.27.163:3000") parser.add_argument("--repo", default="Timmy_Foundation/timmy-home") parser.add_argument("--poll-interval", type=int, default=60) parser.add_argument("--no-timmy-approval", action="store_true", help="Skip Timmy review phase") parser.add_argument("--author-whitelist", help="Comma-separated list of authorized Gitea usernames") parser.add_argument("--no-author-whitelist", action="store_true", help="Disable author whitelist enforcement (NOT RECOMMENDED)") args = parser.parse_args() # Parse whitelist from command line or environment whitelist = None if args.author_whitelist: whitelist = [u.strip() for u in args.author_whitelist.split(",") if u.strip()] elif os.environ.get("TIMMY_AUTHOR_WHITELIST"): whitelist = [u.strip() for u in os.environ.get("TIMMY_AUTHOR_WHITELIST").split(",") if u.strip()] router = ThreeHouseTaskRouter( gitea_url=args.gitea_url, repo=args.repo, poll_interval=args.poll_interval, require_timmy_approval=not args.no_timmy_approval, author_whitelist=whitelist, enforce_author_whitelist=not args.no_author_whitelist ) try: router.start() except KeyboardInterrupt: router.stop() if __name__ == "__main__": main()