#!/usr/bin/env python3 """Task Gate — Pre-task and post-task quality gates for fleet agents. This is the missing enforcement layer between the orchestrator dispatching an issue and an agent submitting a PR. SOUL.md demands "grounding before generation" and "the apparatus that gives these words teeth" — this script is that apparatus. Usage: python3 task_gate.py pre --repo timmy-config --issue 123 --agent groq python3 task_gate.py post --repo timmy-config --issue 123 --agent groq --branch groq/issue-123 Pre-task gate checks: 1. Issue is not already assigned to a different agent 2. No existing branch targets this issue 3. No open PR already addresses this issue 4. Agent is in the correct lane per playbooks/agent-lanes.json 5. Issue is not filtered (epic, permanent, etc.) Post-task gate checks: 1. Branch exists and has commits ahead of main 2. Changed files pass syntax_guard.py 3. No duplicate PR exists for the same issue 4. Branch name follows convention: {agent}/{description} 5. At least one file was actually changed Exit codes: 0 = all gates pass 1 = gate failure (should not proceed) 2 = warning (can proceed with caution) """ import argparse import json import os import subprocess import sys import urllib.request import urllib.error # --------------------------------------------------------------------------- # CONFIG # --------------------------------------------------------------------------- GITEA_API = "https://forge.alexanderwhitestone.com/api/v1" GITEA_OWNER = "Timmy_Foundation" FILTER_TAGS = ["[EPIC]", "[DO NOT CLOSE]", "[PERMANENT]", "[PHILOSOPHY]", "[MORNING REPORT]"] AGENT_USERNAMES = { "groq", "ezra", "bezalel", "allegro", "timmy", "thetimmyc", "perplexity", "kimiclaw", "codex-agent", "manus", "claude", "gemini", "grok", } # --------------------------------------------------------------------------- # GITEA API # --------------------------------------------------------------------------- def load_gitea_token(): token = os.environ.get("GITEA_TOKEN", "") if token: return token.strip() for path in [ os.path.expanduser("~/.hermes/gitea_token_vps"), os.path.expanduser("~/.hermes/gitea_token"), ]: try: with open(path) as f: return f.read().strip() except FileNotFoundError: continue print("[FATAL] No GITEA_TOKEN found") sys.exit(2) def gitea_get(path): token = load_gitea_token() url = f"{GITEA_API}{path}" req = urllib.request.Request(url, headers={ "Authorization": f"token {token}", "Accept": "application/json", }) try: with urllib.request.urlopen(req, timeout=15) as resp: return json.loads(resp.read().decode()) except urllib.error.HTTPError as e: if e.code == 404: return None print(f"[API ERROR] {url} -> {e.code}") return None except Exception as e: print(f"[API ERROR] {url} -> {e}") return None # --------------------------------------------------------------------------- # LANE CHECKER # --------------------------------------------------------------------------- def load_agent_lanes(): """Load agent lane assignments from playbooks/agent-lanes.json.""" lanes_path = os.path.join( os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "playbooks", "agent-lanes.json" ) try: with open(lanes_path) as f: return json.load(f) except FileNotFoundError: return {} # no lanes file = no lane enforcement def check_agent_lane(agent, issue_title, issue_labels, lanes): """Check if the agent is in the right lane for this issue type.""" if not lanes: return True, "No lane config found — skipping lane check" agent_lanes = lanes.get(agent, []) if not agent_lanes: return True, f"No lanes defined for {agent} — skipping" # This is advisory, not blocking — return warning if mismatch return True, f"{agent} has lanes: {agent_lanes}" # --------------------------------------------------------------------------- # PRE-TASK GATE # --------------------------------------------------------------------------- def pre_task_gate(repo, issue_number, agent): """Run all pre-task checks. Returns (pass, messages).""" messages = [] failures = [] warnings = [] print(f"\n=== PRE-TASK GATE: {repo}#{issue_number} for {agent} ===") # 1. Fetch issue issue = gitea_get(f"/repos/{GITEA_OWNER}/{repo}/issues/{issue_number}") if not issue: failures.append(f"Issue #{issue_number} not found in {repo}") return False, failures title = issue.get("title", "") print(f" Issue: {title}") # 2. Check if filtered title_upper = title.upper() for tag in FILTER_TAGS: if tag.upper().replace("[", "").replace("]", "") in title_upper: failures.append(f"Issue has filter tag: {tag} — should not be auto-dispatched") # 3. Check assignees assignees = [a.get("login", "") for a in (issue.get("assignees") or [])] other_agents = [a for a in assignees if a.lower() in AGENT_USERNAMES and a.lower() != agent.lower()] if other_agents: failures.append(f"Already assigned to other agent(s): {other_agents}") # 4. Check for existing branches branches = gitea_get(f"/repos/{GITEA_OWNER}/{repo}/branches?limit=50") if branches: issue_branches = [ b["name"] for b in branches if str(issue_number) in b.get("name", "") and b["name"] != "main" ] if issue_branches: warnings.append(f"Existing branches may target this issue: {issue_branches}") # 5. Check for existing PRs prs = gitea_get(f"/repos/{GITEA_OWNER}/{repo}/pulls?state=open&limit=50") if prs: issue_prs = [ f"PR #{p['number']}: {p['title']}" for p in prs if str(issue_number) in p.get("title", "") or str(issue_number) in p.get("body", "") ] if issue_prs: failures.append(f"Open PR(s) already target this issue: {issue_prs}") # 6. Check agent lanes lanes = load_agent_lanes() labels = [l.get("name", "") for l in (issue.get("labels") or [])] lane_ok, lane_msg = check_agent_lane(agent, title, labels, lanes) if not lane_ok: warnings.append(lane_msg) else: messages.append(f" Lane: {lane_msg}") # Report if failures: print("\n FAILURES:") for f in failures: print(f" ❌ {f}") if warnings: print("\n WARNINGS:") for w in warnings: print(f" ⚠️ {w}") if not failures and not warnings: print(" \u2705 All pre-task gates passed") passed = len(failures) == 0 return passed, failures + warnings # --------------------------------------------------------------------------- # POST-TASK GATE # --------------------------------------------------------------------------- def post_task_gate(repo, issue_number, agent, branch): """Run all post-task checks. Returns (pass, messages).""" failures = [] warnings = [] print(f"\n=== POST-TASK GATE: {repo}#{issue_number} by {agent} ===") print(f" Branch: {branch}") # 1. Check branch exists branch_info = gitea_get( f"/repos/{GITEA_OWNER}/{repo}/branches/{urllib.parse.quote(branch, safe='')}" ) if not branch_info: failures.append(f"Branch '{branch}' does not exist") return False, failures # 2. Check branch naming convention if "/" not in branch: warnings.append(f"Branch name '{branch}' doesn't follow agent/description convention") elif not branch.startswith(f"{agent}/"): warnings.append(f"Branch '{branch}' doesn't start with agent name '{agent}/") # 3. Check for commits ahead of main compare = gitea_get( f"/repos/{GITEA_OWNER}/{repo}/compare/main...{urllib.parse.quote(branch, safe='')}" ) if compare: commits = compare.get("commits", []) if not commits: failures.append("Branch has no commits ahead of main") else: print(f" Commits ahead: {len(commits)}") files = compare.get("diff_files", []) or [] if not files: # Try alternate key num_files = compare.get("total_commits", 0) print(f" Files changed: (check PR diff)") else: print(f" Files changed: {len(files)}") # 4. Check for duplicate PRs prs = gitea_get(f"/repos/{GITEA_OWNER}/{repo}/pulls?state=open&limit=50") if prs: dupe_prs = [ f"PR #{p['number']}" for p in prs if str(issue_number) in p.get("title", "") or str(issue_number) in p.get("body", "") ] if len(dupe_prs) > 1: warnings.append(f"Multiple open PRs may target issue #{issue_number}: {dupe_prs}") # 5. Run syntax guard on changed files (if available) syntax_guard = os.path.join( os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "hermes-sovereign", "scripts", "syntax_guard.py" ) if os.path.exists(syntax_guard): try: result = subprocess.run( [sys.executable, syntax_guard], capture_output=True, text=True, timeout=30 ) if result.returncode != 0: failures.append(f"Syntax guard failed: {result.stdout[:200]}") else: print(" Syntax guard: passed") except Exception as e: warnings.append(f"Could not run syntax guard: {e}") else: warnings.append("syntax_guard.py not found — skipping syntax check") # Report if failures: print("\n FAILURES:") for f in failures: print(f" ❌ {f}") if warnings: print("\n WARNINGS:") for w in warnings: print(f" ⚠️ {w}") if not failures and not warnings: print(" \u2705 All post-task gates passed") passed = len(failures) == 0 return passed, failures + warnings # --------------------------------------------------------------------------- # MAIN # --------------------------------------------------------------------------- def main(): parser = argparse.ArgumentParser(description="Task Gate — pre/post-task quality gates") subparsers = parser.add_subparsers(dest="command") # Pre-task pre = subparsers.add_parser("pre", help="Run pre-task gates") pre.add_argument("--repo", required=True) pre.add_argument("--issue", type=int, required=True) pre.add_argument("--agent", required=True) # Post-task post = subparsers.add_parser("post", help="Run post-task gates") post.add_argument("--repo", required=True) post.add_argument("--issue", type=int, required=True) post.add_argument("--agent", required=True) post.add_argument("--branch", required=True) args = parser.parse_args() if not args.command: parser.print_help() sys.exit(1) if args.command == "pre": passed, msgs = pre_task_gate(args.repo, args.issue, args.agent) elif args.command == "post": passed, msgs = post_task_gate(args.repo, args.issue, args.agent, args.branch) else: parser.print_help() sys.exit(1) sys.exit(0 if passed else 1) if __name__ == "__main__": main()