""" Repository для Governance, Audit, Incidents (PostgreSQL) Використовує таблиці: agents, event_outbox, incidents, incident_history, permissions, agent_revocations """ import json import logging from typing import Optional, List, Dict, Any from datetime import datetime import secrets from repo_city import get_pool, generate_id logger = logging.getLogger(__name__) # ============================================================================= # Governance Repository # ============================================================================= async def get_city_governance_agents() -> List[dict]: """Отримати City Governance агентів (DAARWIZZ, DARIO, DARIA)""" pool = await get_pool() query = """ SELECT a.id, a.display_name, a.kind, a.avatar_url, a.status, a.gov_level, a.node_id, a.created_at FROM agents a WHERE a.gov_level = 'city_governance' AND COALESCE(a.is_archived, false) = false AND a.deleted_at IS NULL ORDER BY a.display_name """ rows = await pool.fetch(query) return [dict(row) for row in rows] async def get_district_agents(district_id: str) -> List[dict]: """Отримати агентів дистрикту (District Lead + core-team)""" pool = await get_pool() query = """ SELECT a.id, a.display_name, a.kind, a.avatar_url, a.status, a.gov_level, aa.role, aa.scope_type, a.created_at FROM agents a LEFT JOIN agent_assignments aa ON a.id = aa.agent_id AND aa.scope_type = 'district' AND aa.scope_id = $1 AND aa.status = 'active' WHERE ( a.gov_level = 'district_lead' OR aa.id IS NOT NULL ) AND COALESCE(a.is_archived, false) = false AND a.deleted_at IS NULL ORDER BY CASE a.gov_level WHEN 'district_lead' THEN 0 ELSE 1 END, a.display_name """ rows = await pool.fetch(query, district_id) return [dict(row) for row in rows] async def get_microdao_agents(microdao_id: str) -> List[dict]: """Отримати агентів MicroDAO (Orchestrator + workers)""" pool = await get_pool() query = """ SELECT a.id, a.display_name, a.kind, a.avatar_url, a.status, a.gov_level, ma.role, ma.is_core, a.created_at FROM agents a JOIN microdao_agents ma ON a.id = ma.agent_id AND ma.microdao_id = $1 WHERE COALESCE(a.is_archived, false) = false AND a.deleted_at IS NULL ORDER BY ma.is_core DESC, CASE a.gov_level WHEN 'orchestrator' THEN 0 WHEN 'core_team' THEN 1 WHEN 'worker' THEN 2 ELSE 3 END, a.display_name """ rows = await pool.fetch(query, microdao_id) return [dict(row) for row in rows] async def get_agents_by_level(level: str) -> List[dict]: """Отримати агентів за gov_level""" pool = await get_pool() query = """ SELECT a.id, a.display_name, a.kind, a.avatar_url, a.status, a.gov_level, a.node_id, a.created_at FROM agents a WHERE a.gov_level = $1 AND COALESCE(a.is_archived, false) = false AND a.deleted_at IS NULL ORDER BY a.display_name """ rows = await pool.fetch(query, level) return [dict(row) for row in rows] async def get_agent_roles(agent_id: str) -> dict: """Отримати ролі та повноваження агента""" pool = await get_pool() # Get agent info agent_query = """ SELECT id, display_name, gov_level, status, home_microdao_id, dais_identity_id FROM agents WHERE id = $1 """ agent = await pool.fetchrow(agent_query, agent_id) if not agent: return {"error": "Agent not found"} # Get assignments assignments_query = """ SELECT id, scope_type, scope_id, role, status, created_at FROM agent_assignments WHERE agent_id = $1 AND status = 'active' ORDER BY created_at DESC """ assignments = await pool.fetch(assignments_query, agent_id) # Get permissions permissions_query = """ SELECT action, target, scope_type, scope_id, granted_by, granted_at FROM permissions WHERE agent_id = $1 ORDER BY granted_at DESC """ permissions = await pool.fetch(permissions_query, agent_id) return { "agent": dict(agent), "assignments": [dict(a) for a in assignments], "permissions": [dict(p) for p in permissions] } async def promote_agent( agent_id: str, new_level: str, actor_id: str ) -> Optional[dict]: """Підвищити агента до нового рівня""" pool = await get_pool() async with pool.acquire() as conn: async with conn.transaction(): # Update agent update_query = """ UPDATE agents SET gov_level = $2, updated_at = NOW() WHERE id = $1 RETURNING id, display_name, gov_level, status """ agent = await conn.fetchrow(update_query, agent_id, new_level) if not agent: return None # Log event await log_governance_event( conn=conn, event_type="agent.promoted", actor_id=actor_id, target_id=agent_id, scope="city", payload={"new_level": new_level} ) return dict(agent) async def demote_agent( agent_id: str, new_level: str, actor_id: str ) -> Optional[dict]: """Понизити рівень агента""" pool = await get_pool() async with pool.acquire() as conn: async with conn.transaction(): # Update agent update_query = """ UPDATE agents SET gov_level = $2, updated_at = NOW() WHERE id = $1 RETURNING id, display_name, gov_level, status """ agent = await conn.fetchrow(update_query, agent_id, new_level) if not agent: return None # Log event await log_governance_event( conn=conn, event_type="agent.demoted", actor_id=actor_id, target_id=agent_id, scope="city", payload={"new_level": new_level} ) return dict(agent) async def revoke_agent( agent_id: str, actor_id: str, reason: str, revocation_type: str = "soft" ) -> Optional[dict]: """Відкликати агента (soft/hard)""" pool = await get_pool() async with pool.acquire() as conn: async with conn.transaction(): # Update agent status new_status = "revoked" if revocation_type == "hard" else "suspended" update_query = """ UPDATE agents SET status = $2, revoked_at = NOW(), revoked_by = $3, updated_at = NOW() WHERE id = $1 RETURNING id, display_name, gov_level, status """ agent = await conn.fetchrow(update_query, agent_id, new_status, actor_id) if not agent: return None # Record revocation revocation_query = """ INSERT INTO agent_revocations ( id, agent_id, revoked_by, reason, revocation_type, created_at ) VALUES ($1, $2, $3, $4, $5, NOW()) """ rev_id = generate_id("rev") await conn.execute(revocation_query, rev_id, agent_id, actor_id, reason, revocation_type) # Revoke DAIS keys if hard revocation if revocation_type == "hard": await conn.execute(""" UPDATE dais_keys SET revoked = true, revoked_reason = $2, revoked_by = $3 WHERE dais_identity_id = ( SELECT dais_identity_id FROM agents WHERE id = $1 ) """, agent_id, reason, actor_id) # Log event await log_governance_event( conn=conn, event_type="agent.revoked", actor_id=actor_id, target_id=agent_id, scope="city", payload={"reason": reason, "type": revocation_type} ) return dict(agent) async def suspend_agent( agent_id: str, actor_id: str, reason: str ) -> Optional[dict]: """Тимчасово призупинити агента""" pool = await get_pool() async with pool.acquire() as conn: async with conn.transaction(): update_query = """ UPDATE agents SET status = 'suspended', updated_at = NOW() WHERE id = $1 RETURNING id, display_name, gov_level, status """ agent = await conn.fetchrow(update_query, agent_id) if not agent: return None await log_governance_event( conn=conn, event_type="agent.suspended", actor_id=actor_id, target_id=agent_id, scope="city", payload={"reason": reason} ) return dict(agent) async def reinstate_agent( agent_id: str, actor_id: str ) -> Optional[dict]: """Відновити призупиненого агента""" pool = await get_pool() async with pool.acquire() as conn: async with conn.transaction(): update_query = """ UPDATE agents SET status = 'online', revoked_at = NULL, revoked_by = NULL, updated_at = NOW() WHERE id = $1 RETURNING id, display_name, gov_level, status """ agent = await conn.fetchrow(update_query, agent_id) if not agent: return None await log_governance_event( conn=conn, event_type="agent.reinstated", actor_id=actor_id, target_id=agent_id, scope="city", payload={} ) return dict(agent) async def check_permission( agent_id: str, action: str, target: str, scope_type: Optional[str] = None, scope_id: Optional[str] = None ) -> dict: """Перевірити чи має агент право на дію""" pool = await get_pool() # Get agent gov_level agent = await pool.fetchrow( "SELECT gov_level, status FROM agents WHERE id = $1", agent_id ) if not agent: return {"allowed": False, "reason": "Agent not found"} if agent["status"] in ("suspended", "revoked"): return {"allowed": False, "reason": f"Agent is {agent['status']}"} gov_level = agent["gov_level"] # City governance can do anything if gov_level == "city_governance": return {"allowed": True, "reason": "City governance has full access"} # Check explicit permission permission_query = """ SELECT 1 FROM permissions WHERE agent_id = $1 AND action = $2 AND target = $3 AND ($4::text IS NULL OR scope_type = $4) AND ($5::text IS NULL OR scope_id = $5) """ has_permission = await pool.fetchrow( permission_query, agent_id, action, target, scope_type, scope_id ) if has_permission: return {"allowed": True, "reason": "Explicit permission granted"} # Check role-based permissions role_permissions = { "orchestrator": ["create_room", "create_task", "manage_members", "moderate"], "core_team": ["create_task", "moderate"], "worker": ["create_task"], "district_lead": ["create_room", "manage_district", "moderate"], "member": [], "guest": [] } allowed_actions = role_permissions.get(gov_level, []) if action in allowed_actions: return {"allowed": True, "reason": f"Allowed by role {gov_level}"} return {"allowed": False, "reason": f"Role {gov_level} cannot perform {action}"} # ============================================================================= # Audit Repository # ============================================================================= async def get_audit_events( limit: int = 50, offset: int = 0, event_type: Optional[str] = None, actor_id: Optional[str] = None, target_id: Optional[str] = None, scope: Optional[str] = None ) -> List[dict]: """Отримати події аудиту з фільтрами""" pool = await get_pool() params = [] where_clauses = [] if event_type: params.append(event_type) where_clauses.append(f"event_type = ${len(params)}") if actor_id: params.append(actor_id) where_clauses.append(f"actor_id = ${len(params)}") if target_id: params.append(target_id) where_clauses.append(f"target_id = ${len(params)}") if scope: params.append(scope) where_clauses.append(f"scope = ${len(params)}") where_sql = " AND ".join(where_clauses) if where_clauses else "1=1" query = f""" SELECT id, event_type, payload, status, created_at, actor_id, target_id, scope FROM event_outbox WHERE {where_sql} ORDER BY created_at DESC LIMIT ${len(params) + 1} OFFSET ${len(params) + 2} """ params.extend([limit, offset]) rows = await pool.fetch(query, *params) return [dict(row) for row in rows] async def get_audit_event_by_id(event_id: str) -> Optional[dict]: """Отримати подію аудиту за ID""" pool = await get_pool() query = """ SELECT id, event_type, payload, status, created_at, actor_id, target_id, scope FROM event_outbox WHERE id = $1 """ row = await pool.fetchrow(query, event_id) return dict(row) if row else None async def get_audit_events_by_actor(actor_id: str, limit: int = 50) -> List[dict]: """Отримати події по актору""" return await get_audit_events(limit=limit, actor_id=actor_id) async def get_audit_events_by_target(target_id: str, limit: int = 50) -> List[dict]: """Отримати події по таргету""" return await get_audit_events(limit=limit, target_id=target_id) async def get_audit_stats() -> dict: """Отримати статистику аудиту""" pool = await get_pool() stats_query = """ SELECT COUNT(*) as total_events, COUNT(CASE WHEN created_at > NOW() - INTERVAL '24 hours' THEN 1 END) as events_24h, COUNT(CASE WHEN created_at > NOW() - INTERVAL '7 days' THEN 1 END) as events_7d, COUNT(DISTINCT actor_id) as unique_actors, COUNT(DISTINCT target_id) as unique_targets FROM event_outbox """ stats = await pool.fetchrow(stats_query) # Top event types types_query = """ SELECT event_type, COUNT(*) as count FROM event_outbox GROUP BY event_type ORDER BY count DESC LIMIT 10 """ types = await pool.fetch(types_query) return { "total_events": stats["total_events"], "events_24h": stats["events_24h"], "events_7d": stats["events_7d"], "unique_actors": stats["unique_actors"], "unique_targets": stats["unique_targets"], "top_event_types": [dict(t) for t in types] } async def log_governance_event( conn, event_type: str, actor_id: str, target_id: str, scope: str, payload: dict ) -> str: """Записати governance подію в event_outbox""" event_id = generate_id("evt") query = """ INSERT INTO event_outbox ( id, event_type, payload, status, actor_id, target_id, scope, created_at ) VALUES ($1, $2, $3::jsonb, 'pending', $4, $5, $6, NOW()) RETURNING id """ result = await conn.fetchrow( query, event_id, event_type, json.dumps(payload), actor_id, target_id, scope ) return result["id"] if result else event_id # ============================================================================= # Incidents Repository # ============================================================================= async def get_incidents( limit: int = 50, offset: int = 0, status: Optional[str] = None, priority: Optional[str] = None, scope_type: Optional[str] = None, scope_id: Optional[str] = None ) -> List[dict]: """Отримати інциденти з фільтрами""" pool = await get_pool() params = [] where_clauses = [] if status: params.append(status) where_clauses.append(f"i.status = ${len(params)}") if priority: params.append(priority) where_clauses.append(f"i.priority = ${len(params)}") if scope_type: params.append(scope_type) where_clauses.append(f"i.target_scope_type = ${len(params)}") if scope_id: params.append(scope_id) where_clauses.append(f"i.target_scope_id = ${len(params)}") where_sql = " AND ".join(where_clauses) if where_clauses else "1=1" query = f""" SELECT i.id::text as id, i.title, i.description, i.status::text, i.priority::text, i.target_scope_type::text as scope_type, i.target_scope_id as scope_id, i.escalation_level::text, i.created_by_dais_id as reporter_id, i.assigned_to_dais_id as assigned_to, i.created_at, i.updated_at, r.display_name as reporter_name, a.display_name as assignee_name FROM incidents i LEFT JOIN agents r ON i.created_by_dais_id = r.id LEFT JOIN agents a ON i.assigned_to_dais_id = a.id WHERE {where_sql} ORDER BY CASE i.priority WHEN 'critical' THEN 0 WHEN 'high' THEN 1 WHEN 'medium' THEN 2 ELSE 3 END, i.created_at DESC LIMIT ${len(params) + 1} OFFSET ${len(params) + 2} """ params.extend([limit, offset]) rows = await pool.fetch(query, *params) return [dict(row) for row in rows] async def get_incident_by_id(incident_id: str) -> Optional[dict]: """Отримати інцидент за ID""" pool = await get_pool() import uuid try: incident_uuid = uuid.UUID(incident_id) except ValueError: return None query = """ SELECT i.id::text as id, i.title, i.description, i.status::text, i.priority::text, i.target_scope_type::text as scope_type, i.target_scope_id as scope_id, i.escalation_level::text, i.created_by_dais_id as reporter_id, i.assigned_to_dais_id as assigned_to, i.created_at, i.updated_at, r.display_name as reporter_name, a.display_name as assignee_name FROM incidents i LEFT JOIN agents r ON i.created_by_dais_id = r.id LEFT JOIN agents a ON i.assigned_to_dais_id = a.id WHERE i.id = $1 """ row = await pool.fetchrow(query, incident_uuid) return dict(row) if row else None async def create_incident( title: str, description: str, reporter_id: str, priority: str = "medium", scope_type: Optional[str] = None, scope_id: Optional[str] = None ) -> dict: """Створити новий інцидент""" pool = await get_pool() # Default scope if not provided if not scope_type: scope_type = "city" if not scope_id: scope_id = "daarion" query = """ INSERT INTO incidents ( title, description, status, priority, target_scope_type, target_scope_id, escalation_level, created_by_dais_id, created_at, updated_at ) VALUES ($1, $2, 'open', $3::incident_priority, $4::target_scope_type, $5, 'microdao', $6, NOW(), NOW()) RETURNING id::text, title, status::text, priority::text, created_at """ row = await pool.fetchrow( query, title, description, priority, scope_type, scope_id, reporter_id ) if row: # Log to history await add_incident_history(row["id"], reporter_id, "created", {"title": title}) return dict(row) if row else {} async def update_incident_status( incident_id: str, new_status: str, actor_id: str ) -> Optional[dict]: """Оновити статус інциденту""" pool = await get_pool() import uuid try: incident_uuid = uuid.UUID(incident_id) except ValueError: return None query = """ UPDATE incidents SET status = $2::incident_status, updated_at = NOW() WHERE id = $1 RETURNING id::text, title, status::text, priority::text, updated_at """ row = await pool.fetchrow(query, incident_uuid, new_status) if row: await add_incident_history(incident_id, actor_id, "status_changed", {"new_status": new_status}) return dict(row) if row else None async def assign_incident( incident_id: str, assignee_id: str, actor_id: str ) -> Optional[dict]: """Призначити інцидент на агента""" pool = await get_pool() import uuid try: incident_uuid = uuid.UUID(incident_id) except ValueError: return None query = """ UPDATE incidents SET assigned_to_dais_id = $2, status = 'in_progress', updated_at = NOW() WHERE id = $1 RETURNING id::text, title, status::text, assigned_to_dais_id as assigned_to, updated_at """ row = await pool.fetchrow(query, incident_uuid, assignee_id) if row: await add_incident_history(incident_id, actor_id, "assigned", {"assignee_id": assignee_id}) return dict(row) if row else None async def escalate_incident( incident_id: str, actor_id: str ) -> Optional[dict]: """Ескалювати інцидент на вищий рівень""" pool = await get_pool() import uuid try: incident_uuid = uuid.UUID(incident_id) except ValueError: return None # Get current level current = await pool.fetchrow( "SELECT escalation_level::text FROM incidents WHERE id = $1", incident_uuid ) if not current: return None # Define escalation path escalation_path = { "microdao": "district", "district": "city", "city": "city" # Max level } new_level = escalation_path.get(current["escalation_level"], "city") query = """ UPDATE incidents SET escalation_level = $2::escalation_level, updated_at = NOW() WHERE id = $1 RETURNING id::text, title, status::text, escalation_level::text, updated_at """ row = await pool.fetchrow(query, incident_uuid, new_level) if row: await add_incident_history(incident_id, actor_id, "escalated", {"new_level": new_level}) return dict(row) if row else None async def resolve_incident( incident_id: str, actor_id: str, resolution: str ) -> Optional[dict]: """Вирішити інцидент""" pool = await get_pool() import uuid try: incident_uuid = uuid.UUID(incident_id) except ValueError: return None query = """ UPDATE incidents SET status = 'resolved', resolution = $2, resolved_at = NOW(), updated_at = NOW() WHERE id = $1 RETURNING id::text, title, status::text, updated_at """ row = await pool.fetchrow(query, incident_uuid, resolution) if row: await add_incident_history(incident_id, actor_id, "resolved", {"resolution": resolution}) return dict(row) if row else None async def close_incident( incident_id: str, actor_id: str ) -> Optional[dict]: """Закрити інцидент""" pool = await get_pool() import uuid try: incident_uuid = uuid.UUID(incident_id) except ValueError: return None query = """ UPDATE incidents SET status = 'closed', closed_at = NOW(), updated_at = NOW() WHERE id = $1 RETURNING id::text, title, status::text, updated_at """ row = await pool.fetchrow(query, incident_uuid) if row: await add_incident_history(incident_id, actor_id, "closed", {}) return dict(row) if row else None async def add_incident_comment( incident_id: str, actor_id: str, comment: str ) -> dict: """Додати коментар до інциденту""" pool = await get_pool() import uuid try: incident_uuid = uuid.UUID(incident_id) except ValueError: return {} query = """ INSERT INTO incident_history ( incident_id, actor_dais_id, action, comment, created_at ) VALUES ($1, $2, 'comment', $3, NOW()) RETURNING id::text, action, created_at """ row = await pool.fetchrow(query, incident_uuid, actor_id, comment) return dict(row) if row else {} async def add_incident_history( incident_id: str, actor_id: str, action: str, details: dict ) -> dict: """Додати запис в історію інциденту""" pool = await get_pool() import uuid try: incident_uuid = uuid.UUID(incident_id) except ValueError: return {} query = """ INSERT INTO incident_history ( incident_id, actor_dais_id, action, new_value, created_at ) VALUES ($1, $2, $3, $4::jsonb, NOW()) RETURNING id::text, action, created_at """ row = await pool.fetchrow( query, incident_uuid, actor_id, action, json.dumps(details) ) return dict(row) if row else {} async def get_incident_history(incident_id: str) -> List[dict]: """Отримати історію інциденту""" pool = await get_pool() import uuid try: incident_uuid = uuid.UUID(incident_id) except ValueError: return [] query = """ SELECT ih.id::text, ih.action, ih.new_value as details, ih.comment, ih.created_at, a.display_name as actor_name FROM incident_history ih LEFT JOIN agents a ON ih.actor_dais_id = a.id WHERE ih.incident_id = $1 ORDER BY ih.created_at DESC """ rows = await pool.fetch(query, incident_uuid) return [dict(row) for row in rows]