# python/src/server/services/propose_change_service.py import asyncio import logging from pathlib import Path from typing import Any, cast from uuid import UUID import aiofiles from ..utils import get_supabase_client class ActionExecutor: """Handles the actual execution of an approved change.""" async def _run_command(self, *args: str) -> str: process = await asyncio.create_subprocess_exec( *args, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) stdout, stderr = await process.communicate() if process.returncode != 0: logging.error(f"Command failed: {stderr.decode().strip()}") raise RuntimeError("Command failed") return stdout.decode().strip() async def execute_file_change(self, payload: dict[str, Any], task_id: str = "ai-fix") -> str: file_path_str = payload.get("file_path") new_content = payload.get("new_content") if not file_path_str or new_content is None: raise ValueError("Invalid payload") from ..utils.code_modifier import CodeModifier modifier = CodeModifier() # 1. Create a sandbox branch for safety branch_name = modifier.create_sandbox_branch(task_id) # 2. Apply the modification modifier.apply_modification(file_path_str, new_content) return f"File '{file_path_str}' written to branch '{branch_name}'" class ProposeChangeService: def __init__(self, db_client=None): self.db_client = db_client or get_supabase_client() self.executor = ActionExecutor() self.logger = logging.getLogger(__name__) def _resolve_user_id(self, user_id: Any) -> str: """ Dynamically handles both UUIDs and simplified IDs ('1', '2', '3'). Ensures the ID is returned in a format suitable for the database column. """ uid_str = str(user_id) # If it's a simple numeric string, we treat it as a valid identity # but keep it as a string to match the updated TEXT column type. return uid_str async def list_proposals(self, status: str | None = "pending", user_id: str | None = None) -> list[dict[str, Any]]: """Lists proposals, optionally filtered by status and user department scope.""" query = self.db_client.table("proposed_changes").select("*") if status: query = query.eq("status", status) # Physical Department Isolation (Phase 4.6.23 Hardening) if user_id: # First, get the department of the requesting manager p_res = self.db_client.table("profiles").select("department, role").eq("id", user_id).execute() if p_res.data and len(p_res.data) > 0: profile = p_res.data[0] if profile.get("role") != "system_admin": dept = profile.get("department") query = query.filter("request_payload->>created_by_dept", "eq", dept) res = query.order("created_at", desc=True).execute() return cast(list[dict[str, Any]], res.data or []) async def get_proposal(self, proposal_id: UUID) -> dict[str, Any] | None: res = self.db_client.table("proposed_changes").select("*").eq("id", str(proposal_id)).execute() return res.data[0] if res.data else None async def create_file_proposal( self, file_path: str, new_content: str, summary: str, user_id: str | None = None ) -> dict[str, Any]: """Creates a file change proposal, capturing current content as old_content.""" p = Path(file_path) old_content = "" if p.exists() and p.is_file(): async with aiofiles.open(p, encoding="utf-8") as f: old_content = await f.read() # Physical identity embedding (Phase 4.6.23) dept = "General" if user_id: u_res = self.db_client.table("profiles").select("department").eq("id", user_id).execute() dept = u_res.data[0].get("department", "General") if u_res.data else "General" payload = { "file_path": file_path, "old_content": old_content, "new_content": new_content, "created_by": user_id, "created_by_dept": dept, "change_summary": summary, } res = ( self.db_client.table("proposed_changes") .insert({"type": "file", "status": "pending", "request_payload": payload}) .execute() ) return cast(dict[str, Any], res.data[0]) async def create_proposal( self, change_type: str, payload: dict[str, Any], user_id: str | None = None ) -> dict[str, Any]: """Creates a generic proposal (e.g. git commands, feature management) in proposed_changes.""" # Embed physical identity (Phase 4.6.23) dept = "General" if user_id: u_res = self.db_client.table("profiles").select("department").eq("id", user_id).execute() dept = u_res.data[0].get("department", "General") if u_res.data else "General" # Inject created_by and created_by_dept into request_payload for audit request_payload = { **payload, "created_by": user_id, "created_by_dept": dept, } res = ( self.db_client.table("proposed_changes") .insert({"type": change_type, "status": "pending", "request_payload": request_payload}) .execute() ) return cast(dict[str, Any], res.data[0]) async def approve_proposal(self, proposal_id: UUID, user_id: Any) -> dict[str, Any]: resolved_id = self._resolve_user_id(user_id) res = ( self.db_client.table("proposed_changes") .update({"status": "approved", "approved_by": resolved_id, "approved_at": "now()"}) .eq("id", str(proposal_id)) .execute() ) # Physical Execution Trigger (Phase 5.1.3) if res.data: proposal = res.data[0] try: await self.executor.execute_file_change( proposal.get("request_payload", {}), task_id=str(proposal_id)[:8] ) except Exception as e: self.logger.error(f"Failed to execute approved change: {e}") raise # Physical Audit Log (Phase 4.6.41) try: u_res = self.db_client.table("profiles").select("name").eq("id", str(user_id)).execute() user_name = u_res.data[0].get("name", "Unknown Admin") if u_res.data else "Unknown Admin" from .log_service import log_service log_service.create_log_entry( { "project_name": "admin-audit", "user_name": user_name, "gemini_response": f"Proposal {proposal_id} approved by {user_name}", "user_input": f"Approve {proposal_id}", } ) except Exception as e: self.logger.warning(f"Audit log failed: {e}") return cast(dict[str, Any], res.data[0]) async def reject_proposal(self, proposal_id: UUID, user_id: Any) -> dict[str, Any]: resolved_id = self._resolve_user_id(user_id) res = ( self.db_client.table("proposed_changes") .update({"status": "rejected", "approved_by": resolved_id, "approved_at": "now()"}) .eq("id", str(proposal_id)) .execute() ) # Physical Audit Log (Phase 4.6.41) try: u_res = self.db_client.table("profiles").select("name").eq("id", str(user_id)).execute() user_name = u_res.data[0].get("name", "Unknown Admin") if u_res.data else "Unknown Admin" from .log_service import log_service log_service.create_log_entry( { "project_name": "admin-audit", "user_name": user_name, "gemini_response": f"Proposal {proposal_id} rejected by {user_name}", "user_input": f"Reject {proposal_id}", } ) except Exception as e: self.logger.warning(f"Audit log failed: {e}") return cast(dict[str, Any], res.data[0])