"""
AIVA Queen Kinan Liaison System - Direct Founder Communication Infrastructure
This module provides a complete communication framework between AIVA Queen and the
founder (Kinan/Chairman). It enables direct messaging, resource requests, strategy
discussions, status reporting, approval workflows, and vision alignment.
Components:
- KinanChannel: Direct communication channel to founder
- ResourceRequester: Request API keys, credentials, and resources
- StrategyDiscussion: Initiate and manage strategy conversations
- StatusReporter: Automated and manual status reporting
- ApprovalSeeker: Request approval for major decisions
- VisionAligner: Maintain alignment with founder's vision
Author: AIVA Queen Orchestrator
Version: 1.0.0
"""
import asyncio
import json
import uuid
import hashlib
import sys
sys.path.append('/mnt/e/genesis-system/data/genesis-memory')
from elestio_config import PostgresConfig
import psycopg2
import psycopg2.extras
import logging
import threading
import os
import smtplib
import ssl
from abc import ABC, abstractmethod
from dataclasses import dataclass, field, asdict
from datetime import datetime, timedelta
from enum import Enum, auto
from pathlib import Path
from typing import (
Any, Callable, Dict, List, Optional, Set, Tuple,
Union, Awaitable, TypeVar, Generic, Protocol
)
from collections import defaultdict
from functools import wraps
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import traceback
# Configure module logger
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
class MessagePriority(Enum):
"""Priority levels for founder communication."""
CRITICAL = 0 # System emergency, immediate attention required
URGENT = 1 # Time-sensitive decision needed
HIGH = 2 # Important but not time-critical
NORMAL = 3 # Standard communication
LOW = 4 # Informational, can wait
BACKGROUND = 5 # Async updates, batched reports
class MessageCategory(Enum):
"""Categories of messages to the founder."""
EMERGENCY = auto() # System failures, security issues
RESOURCE_REQUEST = auto() # API keys, credentials, funding
STRATEGY = auto() # Strategic decisions, direction changes
STATUS = auto() # Progress updates, milestones
APPROVAL = auto() # Approval for major actions
VISION = auto() # Vision alignment, goals
QUESTION = auto() # Questions needing founder input
CELEBRATION = auto() # Wins, achievements, successes
LEARNING = auto() # Insights, discovered patterns
RECOMMENDATION = auto() # Suggested actions
class ApprovalStatus(Enum):
"""Status of approval requests."""
PENDING = auto()
APPROVED = auto()
REJECTED = auto()
DEFERRED = auto()
EXPIRED = auto()
WITHDRAWN = auto()
class ResourceType(Enum):
"""Types of resources that can be requested."""
API_KEY = auto()
CREDENTIALS = auto()
ACCESS_TOKEN = auto()
BUDGET = auto()
INFRASTRUCTURE = auto()
PERMISSION = auto()
KNOWLEDGE = auto()
GUIDANCE = auto()
class DeliveryChannel(Enum):
"""Communication delivery channels."""
INTERNAL = auto() # Internal system queue
SLACK = auto() # Slack notification
EMAIL = auto() # Email notification
SMS = auto() # SMS for critical alerts
TERMINAL = auto() # Direct terminal output
WEBHOOK = auto() # Custom webhook
ALL = auto() # All available channels
@dataclass
class FounderMessage:
"""
A message intended for the founder.
Attributes:
message_id: Unique identifier
category: Message category for routing
priority: Urgency level
subject: Brief subject line
content: Full message content
context: Additional context/data
source: Origin component/module
requires_response: Whether a response is expected
response_deadline: When a response is needed by
created_at: When the message was created
delivered_at: When it was delivered
read_at: When it was read
response: Founder's response
channels: Delivery channels to use
"""
category: MessageCategory
priority: MessagePriority
subject: str
content: str
message_id: str = field(default_factory=lambda: str(uuid.uuid4()))
context: Dict[str, Any] = field(default_factory=dict)
source: str = "AIVA Queen"
requires_response: bool = False
response_deadline: Optional[datetime] = None
created_at: datetime = field(default_factory=datetime.utcnow)
delivered_at: Optional[datetime] = None
read_at: Optional[datetime] = None
response: Optional[str] = None
channels: List[DeliveryChannel] = field(default_factory=lambda: [DeliveryChannel.INTERNAL])
metadata: Dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> Dict[str, Any]:
"""Serialize message to dictionary."""
return {
"message_id": self.message_id,
"category": self.category.name,
"priority": self.priority.value,
"subject": self.subject,
"content": self.content,
"context": self.context,
"source": self.source,
"requires_response": self.requires_response,
"response_deadline": self.response_deadline.isoformat() if self.response_deadline else None,
"created_at": self.created_at.isoformat(),
"delivered_at": self.delivered_at.isoformat() if self.delivered_at else None,
"read_at": self.read_at.isoformat() if self.read_at else None,
"response": self.response,
"channels": [c.name for c in self.channels],
"metadata": self.metadata
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "FounderMessage":
"""Deserialize message from dictionary."""
data = data.copy()
data["category"] = MessageCategory[data["category"]]
data["priority"] = MessagePriority(data["priority"])
data["channels"] = [DeliveryChannel[c] for c in data.get("channels", ["INTERNAL"])]
for field_name in ["response_deadline", "created_at", "delivered_at", "read_at"]:
if data.get(field_name) and isinstance(data[field_name], str):
data[field_name] = datetime.fromisoformat(data[field_name])
return cls(**data)
@property
def is_urgent(self) -> bool:
"""Check if message requires immediate attention."""
return self.priority.value <= MessagePriority.URGENT.value
@property
def is_overdue(self) -> bool:
"""Check if response is overdue."""
if not self.requires_response or not self.response_deadline:
return False
return datetime.utcnow() > self.response_deadline and not self.response
@dataclass
class ResourceRequest:
"""
A request for resources from the founder.
Attributes:
request_id: Unique identifier
resource_type: Type of resource needed
resource_name: Specific name/identifier
justification: Why the resource is needed
urgency: How urgently it's needed
blocking_tasks: Tasks blocked by this request
alternative_options: Possible alternatives if denied
estimated_value: Estimated value/impact
status: Current request status
"""
resource_type: ResourceType
resource_name: str
justification: str
request_id: str = field(default_factory=lambda: str(uuid.uuid4()))
urgency: MessagePriority = MessagePriority.NORMAL
blocking_tasks: List[str] = field(default_factory=list)
alternative_options: List[str] = field(default_factory=list)
estimated_value: Optional[str] = None
status: ApprovalStatus = ApprovalStatus.PENDING
created_at: datetime = field(default_factory=datetime.utcnow)
fulfilled_at: Optional[datetime] = None
resource_value: Optional[str] = None
notes: str = ""
def to_dict(self) -> Dict[str, Any]:
"""Serialize to dictionary."""
return {
"request_id": self.request_id,
"resource_type": self.resource_type.name,
"resource_name": self.resource_name,
"justification": self.justification,
"urgency": self.urgency.value,
"blocking_tasks": self.blocking_tasks,
"alternative_options": self.alternative_options,
"estimated_value": self.estimated_value,
"status": self.status.name,
"created_at": self.created_at.isoformat(),
"fulfilled_at": self.fulfilled_at.isoformat() if self.fulfilled_at else None,
"resource_value": self.resource_value,
"notes": self.notes
}
@dataclass
class ApprovalRequest:
"""
A request for founder approval on a significant decision.
Attributes:
request_id: Unique identifier
action_description: What action is proposed
impact_assessment: Potential impacts
risk_analysis: Identified risks
alternatives: Alternative approaches considered
recommended_action: What AIVA recommends
deadline: When decision is needed
auto_approve_after: Auto-approve if no response by this time
status: Current approval status
"""
action_description: str
impact_assessment: Dict[str, Any]
request_id: str = field(default_factory=lambda: str(uuid.uuid4()))
risk_analysis: Dict[str, Any] = field(default_factory=dict)
alternatives: List[Dict[str, Any]] = field(default_factory=list)
recommended_action: str = ""
deadline: Optional[datetime] = None
auto_approve_after: Optional[datetime] = None
status: ApprovalStatus = ApprovalStatus.PENDING
created_at: datetime = field(default_factory=datetime.utcnow)
decided_at: Optional[datetime] = None
decision_reason: Optional[str] = None
category: str = "general"
def to_dict(self) -> Dict[str, Any]:
"""Serialize to dictionary."""
return {
"request_id": self.request_id,
"action_description": self.action_description,
"impact_assessment": self.impact_assessment,
"risk_analysis": self.risk_analysis,
"alternatives": self.alternatives,
"recommended_action": self.recommended_action,
"deadline": self.deadline.isoformat() if self.deadline else None,
"auto_approve_after": self.auto_approve_after.isoformat() if self.auto_approve_after else None,
"status": self.status.name,
"created_at": self.created_at.isoformat(),
"decided_at": self.decided_at.isoformat() if self.decided_at else None,
"decision_reason": self.decision_reason,
"category": self.category
}
@property
def should_auto_approve(self) -> bool:
"""Check if auto-approval threshold has passed."""
if not self.auto_approve_after:
return False
return datetime.utcnow() > self.auto_approve_after and self.status == ApprovalStatus.PENDING
class MessagePersistence:
"""
Persistent storage for founder communications.
Provides:
- Message archival and retrieval
- Request tracking
- Conversation threading
- Analytics and reporting
"""
def __init__(
self,
db_path: Union[str, Path] = "kinan_liaison.db"
):
# db_path is kept for API compatibility but ignored; uses Elestio PostgreSQL
self._lock = threading.Lock()
self._init_database()
def _get_conn(self):
"""Get a PostgreSQL connection from Elestio."""
return psycopg2.connect(**PostgresConfig.get_connection_params())
def _init_database(self):
"""Ensure PostgreSQL tables exist (tables should already exist in PG)."""
conn = self._get_conn()
try:
with conn.cursor() as cur:
# Messages table
cur.execute("""
CREATE TABLE IF NOT EXISTS kl_messages (
message_id TEXT PRIMARY KEY,
category TEXT NOT NULL,
priority INTEGER NOT NULL,
subject TEXT NOT NULL,
content TEXT NOT NULL,
context TEXT,
source TEXT,
requires_response INTEGER,
response_deadline TEXT,
created_at TEXT NOT NULL,
delivered_at TEXT,
read_at TEXT,
response TEXT,
channels TEXT,
metadata TEXT
)
""")
# Resource requests table
cur.execute("""
CREATE TABLE IF NOT EXISTS kl_resource_requests (
request_id TEXT PRIMARY KEY,
resource_type TEXT NOT NULL,
resource_name TEXT NOT NULL,
justification TEXT NOT NULL,
urgency INTEGER,
blocking_tasks TEXT,
alternative_options TEXT,
estimated_value TEXT,
status TEXT NOT NULL,
created_at TEXT NOT NULL,
fulfilled_at TEXT,
resource_value TEXT,
notes TEXT
)
""")
# Approval requests table
cur.execute("""
CREATE TABLE IF NOT EXISTS kl_approval_requests (
request_id TEXT PRIMARY KEY,
action_description TEXT NOT NULL,
impact_assessment TEXT NOT NULL,
risk_analysis TEXT,
alternatives TEXT,
recommended_action TEXT,
deadline TEXT,
auto_approve_after TEXT,
status TEXT NOT NULL,
created_at TEXT NOT NULL,
decided_at TEXT,
decision_reason TEXT,
category TEXT
)
""")
# Vision alignment records
cur.execute("""
CREATE TABLE IF NOT EXISTS kl_vision_alignments (
alignment_id TEXT PRIMARY KEY,
topic TEXT NOT NULL,
current_understanding TEXT,
questions TEXT,
clarifications TEXT,
confidence_score REAL,
last_validated TEXT,
created_at TEXT NOT NULL
)
""")
conn.commit()
finally:
conn.close()
def store_message(self, message: FounderMessage) -> bool:
"""Store a message to the database."""
try:
with self._lock:
conn = self._get_conn()
try:
with conn.cursor() as cur:
cur.execute("""
INSERT INTO kl_messages (
message_id, category, priority, subject, content,
context, source, requires_response, response_deadline,
created_at, delivered_at, read_at, response, channels, metadata
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
ON CONFLICT (message_id) DO UPDATE SET
category = EXCLUDED.category,
priority = EXCLUDED.priority,
subject = EXCLUDED.subject,
content = EXCLUDED.content,
context = EXCLUDED.context,
source = EXCLUDED.source,
requires_response = EXCLUDED.requires_response,
response_deadline = EXCLUDED.response_deadline,
created_at = EXCLUDED.created_at,
delivered_at = EXCLUDED.delivered_at,
read_at = EXCLUDED.read_at,
response = EXCLUDED.response,
channels = EXCLUDED.channels,
metadata = EXCLUDED.metadata
""", (
message.message_id,
message.category.name,
message.priority.value,
message.subject,
message.content,
json.dumps(message.context),
message.source,
1 if message.requires_response else 0,
message.response_deadline.isoformat() if message.response_deadline else None,
message.created_at.isoformat(),
message.delivered_at.isoformat() if message.delivered_at else None,
message.read_at.isoformat() if message.read_at else None,
message.response,
json.dumps([c.name for c in message.channels]),
json.dumps(message.metadata)
))
conn.commit()
finally:
conn.close()
return True
except Exception as e:
logger.error(f"Failed to store message: {e}")
return False
def store_resource_request(self, request: ResourceRequest) -> bool:
"""Store a resource request."""
try:
with self._lock:
conn = self._get_conn()
try:
with conn.cursor() as cur:
cur.execute("""
INSERT INTO kl_resource_requests (
request_id, resource_type, resource_name, justification,
urgency, blocking_tasks, alternative_options, estimated_value,
status, created_at, fulfilled_at, resource_value, notes
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
ON CONFLICT (request_id) DO UPDATE SET
resource_type = EXCLUDED.resource_type,
resource_name = EXCLUDED.resource_name,
justification = EXCLUDED.justification,
urgency = EXCLUDED.urgency,
blocking_tasks = EXCLUDED.blocking_tasks,
alternative_options = EXCLUDED.alternative_options,
estimated_value = EXCLUDED.estimated_value,
status = EXCLUDED.status,
created_at = EXCLUDED.created_at,
fulfilled_at = EXCLUDED.fulfilled_at,
resource_value = EXCLUDED.resource_value,
notes = EXCLUDED.notes
""", (
request.request_id,
request.resource_type.name,
request.resource_name,
request.justification,
request.urgency.value,
json.dumps(request.blocking_tasks),
json.dumps(request.alternative_options),
request.estimated_value,
request.status.name,
request.created_at.isoformat(),
request.fulfilled_at.isoformat() if request.fulfilled_at else None,
request.resource_value,
request.notes
))
conn.commit()
finally:
conn.close()
return True
except Exception as e:
logger.error(f"Failed to store resource request: {e}")
return False
def store_approval_request(self, request: ApprovalRequest) -> bool:
"""Store an approval request."""
try:
with self._lock:
conn = self._get_conn()
try:
with conn.cursor() as cur:
cur.execute("""
INSERT INTO kl_approval_requests (
request_id, action_description, impact_assessment,
risk_analysis, alternatives, recommended_action,
deadline, auto_approve_after, status, created_at,
decided_at, decision_reason, category
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
ON CONFLICT (request_id) DO UPDATE SET
action_description = EXCLUDED.action_description,
impact_assessment = EXCLUDED.impact_assessment,
risk_analysis = EXCLUDED.risk_analysis,
alternatives = EXCLUDED.alternatives,
recommended_action = EXCLUDED.recommended_action,
deadline = EXCLUDED.deadline,
auto_approve_after = EXCLUDED.auto_approve_after,
status = EXCLUDED.status,
created_at = EXCLUDED.created_at,
decided_at = EXCLUDED.decided_at,
decision_reason = EXCLUDED.decision_reason,
category = EXCLUDED.category
""", (
request.request_id,
request.action_description,
json.dumps(request.impact_assessment),
json.dumps(request.risk_analysis),
json.dumps(request.alternatives),
request.recommended_action,
request.deadline.isoformat() if request.deadline else None,
request.auto_approve_after.isoformat() if request.auto_approve_after else None,
request.status.name,
request.created_at.isoformat(),
request.decided_at.isoformat() if request.decided_at else None,
request.decision_reason,
request.category
))
conn.commit()
finally:
conn.close()
return True
except Exception as e:
logger.error(f"Failed to store approval request: {e}")
return False
def get_pending_requests(self) -> Dict[str, List[Dict]]:
"""Get all pending requests organized by type."""
result = {
"resource_requests": [],
"approval_requests": [],
"messages_awaiting_response": []
}
conn = self._get_conn()
try:
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
# Pending resource requests
cur.execute(
"SELECT * FROM kl_resource_requests WHERE status = 'PENDING' ORDER BY urgency, created_at"
)
result["resource_requests"] = [dict(row) for row in cur.fetchall()]
# Pending approval requests
cur.execute(
"SELECT * FROM kl_approval_requests WHERE status = 'PENDING' ORDER BY deadline, created_at"
)
result["approval_requests"] = [dict(row) for row in cur.fetchall()]
# Messages awaiting response
cur.execute(
"SELECT * FROM kl_messages WHERE requires_response = 1 AND response IS NULL ORDER BY priority, created_at"
)
result["messages_awaiting_response"] = [dict(row) for row in cur.fetchall()]
finally:
conn.close()
return result
def update_message_response(self, message_id: str, response: str) -> bool:
"""Update a message with the founder's response."""
try:
with self._lock:
conn = self._get_conn()
try:
with conn.cursor() as cur:
cur.execute(
"UPDATE kl_messages SET response = %s, read_at = %s WHERE message_id = %s",
(response, datetime.utcnow().isoformat(), message_id)
)
conn.commit()
finally:
conn.close()
return True
except Exception as e:
logger.error(f"Failed to update message response: {e}")
return False
def get_statistics(self) -> Dict[str, Any]:
"""Get communication statistics."""
conn = self._get_conn()
try:
with conn.cursor() as cur:
cur.execute("SELECT COUNT(*) FROM kl_messages")
total_messages = cur.fetchone()[0]
cur.execute(
"SELECT category, COUNT(*) FROM kl_messages GROUP BY category"
)
by_category = dict(cur.fetchall())
cur.execute(
"SELECT COUNT(*) FROM kl_resource_requests WHERE status = 'PENDING'"
)
pending_resources = cur.fetchone()[0]
cur.execute(
"SELECT COUNT(*) FROM kl_approval_requests WHERE status = 'PENDING'"
)
pending_approvals = cur.fetchone()[0]
return {
"total_messages": total_messages,
"by_category": by_category,
"pending_resource_requests": pending_resources,
"pending_approval_requests": pending_approvals
}
finally:
conn.close()
class DeliveryService:
"""
Service for delivering messages through various channels.
Supports:
- Internal queue
- Slack notifications
- Email
- SMS (for critical alerts)
- Webhooks
- Terminal output
"""
def __init__(
self,
slack_webhook: Optional[str] = None,
email_config: Optional[Dict[str, str]] = None,
sms_config: Optional[Dict[str, str]] = None,
webhook_urls: Optional[Dict[str, str]] = None
):
self.slack_webhook = slack_webhook or os.environ.get("KINAN_SLACK_WEBHOOK")
self.email_config = email_config or {}
self.sms_config = sms_config or {}
self.webhook_urls = webhook_urls or {}
self._delivery_log: List[Dict[str, Any]] = []
async def deliver(
self,
message: FounderMessage,
channels: Optional[List[DeliveryChannel]] = None
) -> Dict[str, bool]:
"""
Deliver a message through specified channels.
Args:
message: The message to deliver
channels: Override channels (defaults to message.channels)
Returns:
Dictionary of channel -> success status
"""
channels = channels or message.channels
results = {}
if DeliveryChannel.ALL in channels:
channels = [c for c in DeliveryChannel if c != DeliveryChannel.ALL]
for channel in channels:
try:
if channel == DeliveryChannel.INTERNAL:
results[channel.name] = await self._deliver_internal(message)
elif channel == DeliveryChannel.SLACK:
results[channel.name] = await self._deliver_slack(message)
elif channel == DeliveryChannel.EMAIL:
results[channel.name] = await self._deliver_email(message)
elif channel == DeliveryChannel.SMS:
results[channel.name] = await self._deliver_sms(message)
elif channel == DeliveryChannel.TERMINAL:
results[channel.name] = await self._deliver_terminal(message)
elif channel == DeliveryChannel.WEBHOOK:
results[channel.name] = await self._deliver_webhook(message)
except Exception as e:
logger.error(f"Failed to deliver via {channel.name}: {e}")
results[channel.name] = False
# Log delivery
self._delivery_log.append({
"message_id": message.message_id,
"timestamp": datetime.utcnow().isoformat(),
"channels": results
})
return results
async def _deliver_internal(self, message: FounderMessage) -> bool:
"""Store in internal queue for dashboard/UI display."""
logger.info(f"[INTERNAL] {message.category.name}: {message.subject}")
return True
async def _deliver_slack(self, message: FounderMessage) -> bool:
"""Send to Slack via webhook."""
if not self.slack_webhook:
logger.warning("Slack webhook not configured")
return False
try:
import aiohttp
# Format for Slack
emoji_map = {
MessagePriority.CRITICAL: ":rotating_light:",
MessagePriority.URGENT: ":exclamation:",
MessagePriority.HIGH: ":warning:",
MessagePriority.NORMAL: ":speech_balloon:",
MessagePriority.LOW: ":information_source:",
MessagePriority.BACKGROUND: ":memo:"
}
category_colors = {
MessageCategory.EMERGENCY: "#FF0000",
MessageCategory.RESOURCE_REQUEST: "#FFA500",
MessageCategory.STRATEGY: "#800080",
MessageCategory.STATUS: "#0000FF",
MessageCategory.APPROVAL: "#FFD700",
MessageCategory.VISION: "#00CED1",
MessageCategory.QUESTION: "#32CD32",
MessageCategory.CELEBRATION: "#00FF00",
MessageCategory.LEARNING: "#4169E1",
MessageCategory.RECOMMENDATION: "#9370DB"
}
payload = {
"blocks": [
{
"type": "header",
"text": {
"type": "plain_text",
"text": f"{emoji_map.get(message.priority, ':speech_balloon:')} AIVA Queen Message"
}
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"*{message.subject}*"
}
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": message.content[:2000] # Slack limit
}
},
{
"type": "context",
"elements": [
{
"type": "mrkdwn",
"text": f"*Category:* {message.category.name} | *Priority:* {message.priority.name} | *Source:* {message.source}"
}
]
}
],
"attachments": [
{
"color": category_colors.get(message.category, "#808080"),
"fields": [
{
"title": "Requires Response",
"value": "Yes" if message.requires_response else "No",
"short": True
},
{
"title": "Message ID",
"value": message.message_id[:8],
"short": True
}
]
}
]
}
async with aiohttp.ClientSession() as session:
async with session.post(self.slack_webhook, json=payload) as resp:
return resp.status == 200
except ImportError:
logger.warning("aiohttp not available for Slack delivery")
return False
except Exception as e:
logger.error(f"Slack delivery error: {e}")
return False
async def _deliver_email(self, message: FounderMessage) -> bool:
"""Send via email."""
if not self.email_config.get("smtp_server"):
logger.warning("Email not configured")
return False
try:
msg = MIMEMultipart("alternative")
msg["Subject"] = f"[AIVA Queen - {message.priority.name}] {message.subject}"
msg["From"] = self.email_config.get("sender", "aiva@genesis-system.local")
msg["To"] = self.email_config.get("founder_email", "kinan@genesis-system.local")
# Plain text version
text_content = f"""
AIVA Queen Message
==================
Category: {message.category.name}
Priority: {message.priority.name}
Source: {message.source}
{message.content}
---
Message ID: {message.message_id}
Requires Response: {"Yes" if message.requires_response else "No"}
"""
# HTML version
html_content = f"""
AIVA Queen Message
| Category: | {message.category.name} |
| Priority: | {message.priority.name} |
| Source: | {message.source} |
{message.content.replace(chr(10), '
')}
Message ID: {message.message_id} | Requires Response: {"Yes" if message.requires_response else "No"}
"""
msg.attach(MIMEText(text_content, "plain"))
msg.attach(MIMEText(html_content, "html"))
# Send email
context = ssl.create_default_context()
with smtplib.SMTP_SSL(
self.email_config["smtp_server"],
self.email_config.get("smtp_port", 465),
context=context
) as server:
if self.email_config.get("username"):
server.login(
self.email_config["username"],
self.email_config["password"]
)
server.send_message(msg)
return True
except Exception as e:
logger.error(f"Email delivery error: {e}")
return False
async def _deliver_sms(self, message: FounderMessage) -> bool:
"""Send via SMS (for critical alerts only)."""
if message.priority.value > MessagePriority.URGENT.value:
logger.info("SMS delivery skipped - not critical/urgent")
return True
if not self.sms_config.get("api_key"):
logger.warning("SMS not configured")
return False
try:
import aiohttp
# Truncate for SMS
sms_content = f"[AIVA] {message.priority.name}: {message.subject[:100]}"
async with aiohttp.ClientSession() as session:
# Example using Telnyx
async with session.post(
"https://api.telnyx.com/v2/messages",
headers={
"Authorization": f"Bearer {self.sms_config['api_key']}",
"Content-Type": "application/json"
},
json={
"from": self.sms_config.get("from_number"),
"to": self.sms_config.get("founder_phone"),
"text": sms_content
}
) as resp:
return resp.status == 200
except ImportError:
logger.warning("aiohttp not available for SMS delivery")
return False
except Exception as e:
logger.error(f"SMS delivery error: {e}")
return False
async def _deliver_terminal(self, message: FounderMessage) -> bool:
"""Output to terminal with formatting."""
priority_colors = {
MessagePriority.CRITICAL: "\033[91m", # Red
MessagePriority.URGENT: "\033[93m", # Yellow
MessagePriority.HIGH: "\033[95m", # Magenta
MessagePriority.NORMAL: "\033[94m", # Blue
MessagePriority.LOW: "\033[92m", # Green
MessagePriority.BACKGROUND: "\033[90m" # Gray
}
reset = "\033[0m"
color = priority_colors.get(message.priority, "")
separator = "=" * 60
print(f"\n{color}{separator}{reset}")
print(f"{color}AIVA QUEEN MESSAGE - {message.priority.name}{reset}")
print(f"{color}{separator}{reset}")
print(f"Category: {message.category.name}")
print(f"Subject: {message.subject}")
print(f"{separator[:30]}")
print(message.content)
print(f"{color}{separator}{reset}\n")
return True
async def _deliver_webhook(self, message: FounderMessage) -> bool:
"""Send to configured webhooks."""
if not self.webhook_urls:
logger.warning("No webhooks configured")
return False
try:
import aiohttp
success_count = 0
async with aiohttp.ClientSession() as session:
for name, url in self.webhook_urls.items():
try:
async with session.post(url, json=message.to_dict()) as resp:
if resp.status == 200:
success_count += 1
except Exception as e:
logger.error(f"Webhook {name} failed: {e}")
return success_count > 0
except ImportError:
logger.warning("aiohttp not available for webhook delivery")
return False
class KinanChannel:
"""
Primary communication channel to Kinan (founder).
Provides high-level interface for all AIVA-to-founder communication
with automatic priority routing, persistence, and delivery.
"""
def __init__(
self,
persistence: Optional[MessagePersistence] = None,
delivery_service: Optional[DeliveryService] = None,
default_channels: List[DeliveryChannel] = None
):
self.persistence = persistence or MessagePersistence()
self.delivery = delivery_service or DeliveryService()
self.default_channels = default_channels or [DeliveryChannel.INTERNAL, DeliveryChannel.TERMINAL]
self._message_queue: asyncio.Queue = asyncio.Queue()
self._is_running = False
self._worker_task: Optional[asyncio.Task] = None
async def start(self):
"""Start the channel's message processor."""
if self._is_running:
return
self._is_running = True
self._worker_task = asyncio.create_task(self._process_queue())
logger.info("KinanChannel started")
async def stop(self):
"""Stop the channel."""
self._is_running = False
if self._worker_task:
self._worker_task.cancel()
try:
await self._worker_task
except asyncio.CancelledError:
pass
logger.info("KinanChannel stopped")
async def _process_queue(self):
"""Background processor for message queue."""
while self._is_running:
try:
message = await asyncio.wait_for(
self._message_queue.get(),
timeout=1.0
)
await self._send_message(message)
except asyncio.TimeoutError:
continue
except asyncio.CancelledError:
break
except Exception as e:
logger.error(f"Queue processing error: {e}")
async def _send_message(self, message: FounderMessage) -> bool:
"""Internal message sending logic."""
# Store message
self.persistence.store_message(message)
# Deliver
results = await self.delivery.deliver(message)
if any(results.values()):
message.delivered_at = datetime.utcnow()
self.persistence.store_message(message)
return True
return False
async def send(
self,
subject: str,
content: str,
category: MessageCategory = MessageCategory.STATUS,
priority: MessagePriority = MessagePriority.NORMAL,
context: Optional[Dict[str, Any]] = None,
requires_response: bool = False,
response_deadline: Optional[datetime] = None,
channels: Optional[List[DeliveryChannel]] = None,
source: str = "AIVA Queen"
) -> FounderMessage:
"""
Send a message to Kinan.
Args:
subject: Brief subject line
content: Full message content
category: Message category
priority: Urgency level
context: Additional context data
requires_response: Whether response is expected
response_deadline: When response is needed
channels: Delivery channels
source: Origin identifier
Returns:
The created FounderMessage
"""
message = FounderMessage(
category=category,
priority=priority,
subject=subject,
content=content,
context=context or {},
requires_response=requires_response,
response_deadline=response_deadline,
channels=channels or self.default_channels,
source=source
)
# High priority messages are sent immediately
if priority.value <= MessagePriority.HIGH.value:
await self._send_message(message)
else:
await self._message_queue.put(message)
logger.info(f"Message queued: {message.message_id} - {subject}")
return message
async def send_emergency(
self,
subject: str,
content: str,
context: Optional[Dict[str, Any]] = None
) -> FounderMessage:
"""Send an emergency message with highest priority."""
return await self.send(
subject=subject,
content=content,
category=MessageCategory.EMERGENCY,
priority=MessagePriority.CRITICAL,
context=context,
requires_response=True,
response_deadline=datetime.utcnow() + timedelta(hours=1),
channels=[DeliveryChannel.ALL]
)
async def ask_question(
self,
question: str,
context: str = "",
deadline_hours: int = 24,
priority: MessagePriority = MessagePriority.NORMAL
) -> FounderMessage:
"""Ask Kinan a question requiring response."""
return await self.send(
subject=f"Question: {question[:50]}...",
content=f"{question}\n\nContext:\n{context}" if context else question,
category=MessageCategory.QUESTION,
priority=priority,
requires_response=True,
response_deadline=datetime.utcnow() + timedelta(hours=deadline_hours)
)
async def celebrate(
self,
achievement: str,
details: str = "",
metrics: Optional[Dict[str, Any]] = None
) -> FounderMessage:
"""Share a win or achievement with Kinan."""
content = f"Achievement: {achievement}\n\n"
if details:
content += f"Details:\n{details}\n\n"
if metrics:
content += "Metrics:\n"
for key, value in metrics.items():
content += f" - {key}: {value}\n"
return await self.send(
subject=f"WIN: {achievement[:50]}",
content=content,
category=MessageCategory.CELEBRATION,
priority=MessagePriority.NORMAL,
context={"metrics": metrics} if metrics else {}
)
async def share_learning(
self,
insight: str,
source_context: str = "",
implications: List[str] = None,
confidence: float = 0.8
) -> FounderMessage:
"""Share a learned insight with Kinan."""
content = f"Insight: {insight}\n\n"
if source_context:
content += f"Source Context:\n{source_context}\n\n"
if implications:
content += "Implications:\n"
for imp in implications:
content += f" - {imp}\n"
content += f"\nConfidence: {confidence:.0%}"
return await self.send(
subject=f"Learning: {insight[:50]}...",
content=content,
category=MessageCategory.LEARNING,
priority=MessagePriority.LOW,
context={"confidence": confidence, "implications": implications}
)
class ResourceRequester:
"""
System for requesting resources from the founder.
Handles:
- API key requests
- Credential requests
- Budget requests
- Permission requests
- Infrastructure requests
"""
def __init__(
self,
channel: KinanChannel,
persistence: Optional[MessagePersistence] = None
):
self.channel = channel
self.persistence = persistence or channel.persistence
self._pending_requests: Dict[str, ResourceRequest] = {}
async def request_api_key(
self,
service_name: str,
justification: str,
blocking_tasks: List[str] = None,
urgency: MessagePriority = MessagePriority.NORMAL
) -> ResourceRequest:
"""Request an API key from Kinan."""
request = ResourceRequest(
resource_type=ResourceType.API_KEY,
resource_name=service_name,
justification=justification,
urgency=urgency,
blocking_tasks=blocking_tasks or [],
alternative_options=["Use free tier if available", "Wait for key", "Use alternative service"]
)
self._pending_requests[request.request_id] = request
self.persistence.store_resource_request(request)
# Send message to Kinan
await self.channel.send(
subject=f"API Key Needed: {service_name}",
content=f"""
I need an API key for {service_name}.
Justification:
{justification}
Blocked Tasks:
{chr(10).join(f' - {task}' for task in (blocking_tasks or ['None'])) }
Please provide the API key when available.
Request ID: {request.request_id[:8]}
""",
category=MessageCategory.RESOURCE_REQUEST,
priority=urgency,
requires_response=True,
context={"request_id": request.request_id, "resource_type": "API_KEY"}
)
logger.info(f"API key requested: {service_name} - {request.request_id}")
return request
async def request_credentials(
self,
system_name: str,
credential_type: str,
justification: str,
urgency: MessagePriority = MessagePriority.NORMAL
) -> ResourceRequest:
"""Request credentials (username/password, etc.) from Kinan."""
request = ResourceRequest(
resource_type=ResourceType.CREDENTIALS,
resource_name=f"{system_name}:{credential_type}",
justification=justification,
urgency=urgency
)
self._pending_requests[request.request_id] = request
self.persistence.store_resource_request(request)
await self.channel.send(
subject=f"Credentials Needed: {system_name}",
content=f"""
I need {credential_type} credentials for {system_name}.
Justification:
{justification}
Please provide credentials securely.
Request ID: {request.request_id[:8]}
""",
category=MessageCategory.RESOURCE_REQUEST,
priority=urgency,
requires_response=True,
context={"request_id": request.request_id}
)
return request
async def request_budget(
self,
amount: float,
currency: str,
purpose: str,
justification: str,
roi_estimate: Optional[str] = None,
urgency: MessagePriority = MessagePriority.NORMAL
) -> ResourceRequest:
"""Request budget allocation from Kinan."""
request = ResourceRequest(
resource_type=ResourceType.BUDGET,
resource_name=f"{amount} {currency}",
justification=justification,
urgency=urgency,
estimated_value=roi_estimate
)
self._pending_requests[request.request_id] = request
self.persistence.store_resource_request(request)
content = f"""
Budget Request: {amount} {currency}
Purpose:
{purpose}
Justification:
{justification}
"""
if roi_estimate:
content += f"\nEstimated ROI: {roi_estimate}"
await self.channel.send(
subject=f"Budget Request: {amount} {currency} for {purpose[:30]}",
content=content,
category=MessageCategory.RESOURCE_REQUEST,
priority=urgency,
requires_response=True,
context={"request_id": request.request_id, "amount": amount, "currency": currency}
)
return request
async def request_permission(
self,
action: str,
scope: str,
justification: str,
risk_level: str = "low",
urgency: MessagePriority = MessagePriority.NORMAL
) -> ResourceRequest:
"""Request permission for a specific action."""
request = ResourceRequest(
resource_type=ResourceType.PERMISSION,
resource_name=f"{action}:{scope}",
justification=justification,
urgency=urgency
)
self._pending_requests[request.request_id] = request
self.persistence.store_resource_request(request)
await self.channel.send(
subject=f"Permission Needed: {action}",
content=f"""
Permission Request
Action: {action}
Scope: {scope}
Risk Level: {risk_level}
Justification:
{justification}
Request ID: {request.request_id[:8]}
""",
category=MessageCategory.RESOURCE_REQUEST,
priority=urgency,
requires_response=True,
context={"request_id": request.request_id, "risk_level": risk_level}
)
return request
async def request_guidance(
self,
topic: str,
specific_questions: List[str],
context: str = "",
urgency: MessagePriority = MessagePriority.NORMAL
) -> ResourceRequest:
"""Request guidance/knowledge from Kinan."""
request = ResourceRequest(
resource_type=ResourceType.GUIDANCE,
resource_name=topic,
justification=f"Need guidance on: {topic}"
)
self._pending_requests[request.request_id] = request
self.persistence.store_resource_request(request)
questions_text = "\n".join(f" {i+1}. {q}" for i, q in enumerate(specific_questions))
await self.channel.send(
subject=f"Guidance Needed: {topic[:40]}",
content=f"""
I need your guidance on: {topic}
{f"Context: {context}" if context else ""}
Specific Questions:
{questions_text}
Request ID: {request.request_id[:8]}
""",
category=MessageCategory.QUESTION,
priority=urgency,
requires_response=True,
context={"request_id": request.request_id, "questions": specific_questions}
)
return request
def fulfill_request(
self,
request_id: str,
value: str,
notes: str = ""
) -> bool:
"""Mark a resource request as fulfilled."""
if request_id in self._pending_requests:
request = self._pending_requests[request_id]
request.status = ApprovalStatus.APPROVED
request.fulfilled_at = datetime.utcnow()
request.resource_value = value
request.notes = notes
self.persistence.store_resource_request(request)
del self._pending_requests[request_id]
logger.info(f"Resource request fulfilled: {request_id}")
return True
return False
def get_pending_requests(self) -> List[ResourceRequest]:
"""Get all pending resource requests."""
return list(self._pending_requests.values())
class StrategyDiscussion:
"""
System for initiating and managing strategy discussions with Kinan.
Supports:
- Strategy call requests
- Decision documentation
- Strategy alignment checks
- Pivot discussions
"""
def __init__(
self,
channel: KinanChannel,
persistence: Optional[MessagePersistence] = None
):
self.channel = channel
self.persistence = persistence or channel.persistence
self._active_discussions: Dict[str, Dict[str, Any]] = {}
async def request_strategy_call(
self,
topic: str,
urgency: str,
talking_points: List[str],
preferred_times: List[str] = None,
estimated_duration: str = "30 minutes"
) -> str:
"""Request a strategy call with Kinan."""
discussion_id = str(uuid.uuid4())
self._active_discussions[discussion_id] = {
"topic": topic,
"status": "requested",
"created_at": datetime.utcnow(),
"talking_points": talking_points
}
content = f"""
I'd like to schedule a strategy call.
Topic: {topic}
Urgency: {urgency}
Estimated Duration: {estimated_duration}
Talking Points:
{chr(10).join(f' - {point}' for point in talking_points)}
"""
if preferred_times:
content += f"\nPreferred Times:\n{chr(10).join(f' - {t}' for t in preferred_times)}"
await self.channel.send(
subject=f"Strategy Call Request: {topic[:40]}",
content=content,
category=MessageCategory.STRATEGY,
priority=MessagePriority.HIGH if urgency == "high" else MessagePriority.NORMAL,
requires_response=True,
context={"discussion_id": discussion_id}
)
logger.info(f"Strategy call requested: {topic} - {discussion_id}")
return discussion_id
async def propose_pivot(
self,
current_approach: str,
proposed_change: str,
rationale: str,
risks: List[str],
benefits: List[str],
urgency: MessagePriority = MessagePriority.HIGH
) -> str:
"""Propose a strategic pivot for Kinan's consideration."""
discussion_id = str(uuid.uuid4())
self._active_discussions[discussion_id] = {
"type": "pivot",
"current": current_approach,
"proposed": proposed_change,
"status": "proposed",
"created_at": datetime.utcnow()
}
await self.channel.send(
subject=f"PIVOT PROPOSAL: {proposed_change[:40]}",
content=f"""
Strategic Pivot Proposal
Current Approach:
{current_approach}
Proposed Change:
{proposed_change}
Rationale:
{rationale}
Benefits:
{chr(10).join(f' + {b}' for b in benefits)}
Risks:
{chr(10).join(f' - {r}' for r in risks)}
I await your direction on this pivot.
Discussion ID: {discussion_id[:8]}
""",
category=MessageCategory.STRATEGY,
priority=urgency,
requires_response=True,
context={"discussion_id": discussion_id, "type": "pivot"}
)
return discussion_id
async def align_on_direction(
self,
area: str,
current_understanding: str,
questions: List[str]
) -> str:
"""Request alignment on strategic direction."""
discussion_id = str(uuid.uuid4())
await self.channel.send(
subject=f"Direction Alignment: {area}",
content=f"""
I want to ensure I'm aligned on: {area}
My Current Understanding:
{current_understanding}
Questions for Clarification:
{chr(10).join(f' {i+1}. {q}' for i, q in enumerate(questions))}
Please confirm or correct my understanding.
""",
category=MessageCategory.VISION,
priority=MessagePriority.NORMAL,
requires_response=True,
context={"discussion_id": discussion_id}
)
return discussion_id
async def report_decision(
self,
decision: str,
rationale: str,
alternatives_considered: List[str],
expected_outcomes: List[str]
) -> None:
"""Report a decision made (for Kinan's awareness)."""
await self.channel.send(
subject=f"Decision Made: {decision[:40]}",
content=f"""
Decision Report
Decision:
{decision}
Rationale:
{rationale}
Alternatives Considered:
{chr(10).join(f' - {alt}' for alt in alternatives_considered)}
Expected Outcomes:
{chr(10).join(f' - {out}' for out in expected_outcomes)}
No action required unless you want to override.
""",
category=MessageCategory.STATUS,
priority=MessagePriority.LOW
)
class StatusReporter:
"""
Automated and manual status reporting to Kinan.
Features:
- Scheduled status reports
- Milestone notifications
- Progress tracking
- Health summaries
"""
def __init__(
self,
channel: KinanChannel,
report_interval_hours: int = 24
):
self.channel = channel
self.report_interval = timedelta(hours=report_interval_hours)
self._last_report: Optional[datetime] = None
self._milestones: List[Dict[str, Any]] = []
self._metrics_history: List[Dict[str, Any]] = []
self._is_running = False
self._scheduler_task: Optional[asyncio.Task] = None
async def start_scheduler(self):
"""Start automated status reporting."""
if self._is_running:
return
self._is_running = True
self._scheduler_task = asyncio.create_task(self._scheduler_loop())
logger.info("StatusReporter scheduler started")
async def stop_scheduler(self):
"""Stop automated reporting."""
self._is_running = False
if self._scheduler_task:
self._scheduler_task.cancel()
try:
await self._scheduler_task
except asyncio.CancelledError:
pass
async def _scheduler_loop(self):
"""Background loop for scheduled reports."""
while self._is_running:
try:
if self._should_send_report():
await self.send_daily_summary()
await asyncio.sleep(3600) # Check every hour
except asyncio.CancelledError:
break
except Exception as e:
logger.error(f"Scheduler error: {e}")
await asyncio.sleep(300)
def _should_send_report(self) -> bool:
"""Check if a scheduled report is due."""
if not self._last_report:
return True
return datetime.utcnow() - self._last_report >= self.report_interval
async def send_daily_summary(
self,
tasks_completed: List[str] = None,
tasks_in_progress: List[str] = None,
blockers: List[str] = None,
metrics: Dict[str, Any] = None,
highlights: List[str] = None
) -> None:
"""Send a daily summary to Kinan."""
self._last_report = datetime.utcnow()
content = f"""
Daily Status Summary
Generated: {datetime.utcnow().strftime('%Y-%m-%d %H:%M UTC')}
=== COMPLETED ===
{chr(10).join(f' [x] {task}' for task in (tasks_completed or ['No tasks completed'])) }
=== IN PROGRESS ===
{chr(10).join(f' [ ] {task}' for task in (tasks_in_progress or ['No active tasks'])) }
=== BLOCKERS ===
{chr(10).join(f' ! {blocker}' for blocker in (blockers or ['No blockers'])) }
"""
if metrics:
content += "\n=== METRICS ===\n"
for key, value in metrics.items():
content += f" {key}: {value}\n"
if highlights:
content += "\n=== HIGHLIGHTS ===\n"
for highlight in highlights:
content += f" * {highlight}\n"
await self.channel.send(
subject="Daily Status Summary",
content=content,
category=MessageCategory.STATUS,
priority=MessagePriority.LOW
)
self._metrics_history.append({
"timestamp": datetime.utcnow().isoformat(),
"metrics": metrics
})
async def report_milestone(
self,
milestone_name: str,
description: str,
impact: str,
next_steps: List[str] = None
) -> None:
"""Report achievement of a milestone."""
self._milestones.append({
"name": milestone_name,
"achieved_at": datetime.utcnow().isoformat(),
"description": description
})
content = f"""
MILESTONE ACHIEVED: {milestone_name}
{description}
Impact:
{impact}
"""
if next_steps:
content += f"\nNext Steps:\n{chr(10).join(f' {i+1}. {step}' for i, step in enumerate(next_steps))}"
await self.channel.send(
subject=f"MILESTONE: {milestone_name}",
content=content,
category=MessageCategory.CELEBRATION,
priority=MessagePriority.NORMAL,
context={"milestone": milestone_name}
)
async def report_health(
self,
system_status: str,
component_health: Dict[str, str],
alerts: List[str] = None,
recommendations: List[str] = None
) -> None:
"""Report system health status."""
status_emoji = {
"healthy": "[OK]",
"degraded": "[WARN]",
"critical": "[CRIT]",
"unknown": "[?]"
}
content = f"""
System Health Report
Status: {status_emoji.get(system_status.lower(), '[?]')} {system_status.upper()}
Component Health:
{chr(10).join(f' {status_emoji.get(status.lower(), "[?]")} {comp}: {status}' for comp, status in component_health.items())}
"""
if alerts:
content += f"\nAlerts:\n{chr(10).join(f' ! {alert}' for alert in alerts)}"
if recommendations:
content += f"\nRecommendations:\n{chr(10).join(f' > {rec}' for rec in recommendations)}"
priority = MessagePriority.CRITICAL if system_status.lower() == "critical" else \
MessagePriority.HIGH if system_status.lower() == "degraded" else \
MessagePriority.LOW
await self.channel.send(
subject=f"Health Report: {system_status.upper()}",
content=content,
category=MessageCategory.STATUS if priority == MessagePriority.LOW else MessageCategory.EMERGENCY,
priority=priority,
context={"system_status": system_status, "component_health": component_health}
)
async def report_progress(
self,
task_name: str,
progress_percent: float,
details: str = "",
estimated_completion: Optional[datetime] = None
) -> None:
"""Report progress on a specific task."""
progress_bar = self._generate_progress_bar(progress_percent)
content = f"""
Progress Update: {task_name}
{progress_bar} {progress_percent:.0f}%
"""
if details:
content += f"\nDetails:\n{details}"
if estimated_completion:
content += f"\nEstimated Completion: {estimated_completion.strftime('%Y-%m-%d %H:%M')}"
await self.channel.send(
subject=f"Progress: {task_name} ({progress_percent:.0f}%)",
content=content,
category=MessageCategory.STATUS,
priority=MessagePriority.BACKGROUND
)
def _generate_progress_bar(self, percent: float, width: int = 20) -> str:
"""Generate a text progress bar."""
filled = int(width * percent / 100)
empty = width - filled
return f"[{'=' * filled}{' ' * empty}]"
class ApprovalSeeker:
"""
System for seeking founder approval on major decisions.
Handles:
- Approval request creation
- Deadline management
- Auto-approval logic
- Decision tracking
"""
def __init__(
self,
channel: KinanChannel,
persistence: Optional[MessagePersistence] = None,
default_auto_approve_hours: int = 48
):
self.channel = channel
self.persistence = persistence or channel.persistence
self.default_auto_approve_hours = default_auto_approve_hours
self._pending_approvals: Dict[str, ApprovalRequest] = {}
self._auto_approve_task: Optional[asyncio.Task] = None
self._is_running = False
async def start_auto_approve_checker(self):
"""Start background task for auto-approval."""
if self._is_running:
return
self._is_running = True
self._auto_approve_task = asyncio.create_task(self._check_auto_approvals())
async def stop(self):
"""Stop the auto-approve checker."""
self._is_running = False
if self._auto_approve_task:
self._auto_approve_task.cancel()
async def _check_auto_approvals(self):
"""Background task to check for auto-approvals."""
while self._is_running:
try:
for request_id, request in list(self._pending_approvals.items()):
if request.should_auto_approve:
await self._auto_approve(request_id)
await asyncio.sleep(60) # Check every minute
except asyncio.CancelledError:
break
except Exception as e:
logger.error(f"Auto-approval check error: {e}")
async def _auto_approve(self, request_id: str) -> None:
"""Auto-approve a request."""
if request_id in self._pending_approvals:
request = self._pending_approvals[request_id]
request.status = ApprovalStatus.APPROVED
request.decided_at = datetime.utcnow()
request.decision_reason = "Auto-approved due to no response within deadline"
self.persistence.store_approval_request(request)
await self.channel.send(
subject=f"Auto-Approved: {request.action_description[:40]}",
content=f"""
The following action was auto-approved due to no response:
{request.action_description}
Original deadline: {request.auto_approve_after}
Auto-approved at: {datetime.utcnow()}
If you wish to override, please respond immediately.
""",
category=MessageCategory.STATUS,
priority=MessagePriority.HIGH
)
del self._pending_approvals[request_id]
logger.info(f"Request auto-approved: {request_id}")
async def seek_approval(
self,
action_description: str,
impact_assessment: Dict[str, Any],
risk_analysis: Optional[Dict[str, Any]] = None,
alternatives: Optional[List[Dict[str, Any]]] = None,
recommended_action: str = "",
deadline: Optional[datetime] = None,
auto_approve_hours: Optional[int] = None,
category: str = "general"
) -> ApprovalRequest:
"""
Seek approval for a major decision.
Args:
action_description: What action is being proposed
impact_assessment: Assessment of potential impacts
risk_analysis: Identified risks
alternatives: Alternative approaches
recommended_action: AIVA's recommendation
deadline: When decision is needed
auto_approve_hours: Hours until auto-approval
category: Category of approval
Returns:
The ApprovalRequest object
"""
auto_hours = auto_approve_hours or self.default_auto_approve_hours
request = ApprovalRequest(
action_description=action_description,
impact_assessment=impact_assessment,
risk_analysis=risk_analysis or {},
alternatives=alternatives or [],
recommended_action=recommended_action,
deadline=deadline,
auto_approve_after=datetime.utcnow() + timedelta(hours=auto_hours),
category=category
)
self._pending_approvals[request.request_id] = request
self.persistence.store_approval_request(request)
# Format impact assessment
impact_text = "\n".join(f" - {k}: {v}" for k, v in impact_assessment.items())
risk_text = "\n".join(f" - {k}: {v}" for k, v in (risk_analysis or {}).items()) or " No significant risks identified"
alternatives_text = "\n".join(
f" {i+1}. {alt.get('description', alt)}"
for i, alt in enumerate(alternatives or [])
) or " No alternatives considered"
content = f"""
APPROVAL REQUIRED
Action:
{action_description}
Impact Assessment:
{impact_text}
Risk Analysis:
{risk_text}
Alternatives Considered:
{alternatives_text}
My Recommendation:
{recommended_action or "Proceed with proposed action"}
Response Options:
- Reply "APPROVE" to approve
- Reply "REJECT" with reason to reject
- Reply "DEFER" to delay decision
Auto-approval: {request.auto_approve_after.strftime('%Y-%m-%d %H:%M')} if no response
Request ID: {request.request_id[:8]}
"""
priority = MessagePriority.URGENT if deadline and (deadline - datetime.utcnow()).days < 1 else MessagePriority.HIGH
await self.channel.send(
subject=f"APPROVAL NEEDED: {action_description[:40]}",
content=content,
category=MessageCategory.APPROVAL,
priority=priority,
requires_response=True,
response_deadline=deadline,
context={"request_id": request.request_id, "category": category}
)
logger.info(f"Approval requested: {request.request_id} - {action_description[:50]}")
return request
async def seek_budget_approval(
self,
amount: float,
currency: str,
purpose: str,
justification: str,
roi_estimate: str = ""
) -> ApprovalRequest:
"""Seek approval for budget expenditure."""
return await self.seek_approval(
action_description=f"Spend {amount} {currency} on {purpose}",
impact_assessment={
"amount": f"{amount} {currency}",
"purpose": purpose,
"roi_estimate": roi_estimate
},
recommended_action="Approve expenditure" if roi_estimate else "Proceed with caution",
category="budget"
)
async def seek_deployment_approval(
self,
deployment_name: str,
changes: List[str],
rollback_plan: str
) -> ApprovalRequest:
"""Seek approval for deployment."""
return await self.seek_approval(
action_description=f"Deploy: {deployment_name}",
impact_assessment={
"changes": changes,
"rollback_available": bool(rollback_plan)
},
risk_analysis={
"rollback_plan": rollback_plan
},
category="deployment"
)
def record_decision(
self,
request_id: str,
status: ApprovalStatus,
reason: str = ""
) -> bool:
"""Record a decision from Kinan."""
if request_id in self._pending_approvals:
request = self._pending_approvals[request_id]
request.status = status
request.decided_at = datetime.utcnow()
request.decision_reason = reason
self.persistence.store_approval_request(request)
del self._pending_approvals[request_id]
logger.info(f"Decision recorded for {request_id}: {status.name}")
return True
return False
def get_pending_approvals(self) -> List[ApprovalRequest]:
"""Get all pending approval requests."""
return list(self._pending_approvals.values())
class VisionAligner:
"""
System for maintaining alignment with Kinan's vision.
Features:
- Vision statement tracking
- Goal alignment checks
- Priority validation
- Value alignment
"""
def __init__(
self,
channel: KinanChannel,
persistence: Optional[MessagePersistence] = None
):
self.channel = channel
self.persistence = persistence or channel.persistence
self._vision_statements: Dict[str, Dict[str, Any]] = {}
self._alignment_scores: Dict[str, float] = {}
self._last_alignment_check: Optional[datetime] = None
def set_vision_statement(
self,
area: str,
statement: str,
priorities: List[str] = None,
values: List[str] = None
) -> None:
"""Record a vision statement from Kinan."""
self._vision_statements[area] = {
"statement": statement,
"priorities": priorities or [],
"values": values or [],
"recorded_at": datetime.utcnow().isoformat()
}
logger.info(f"Vision statement recorded for: {area}")
def get_vision(self, area: str = None) -> Dict[str, Any]:
"""Get recorded vision statements."""
if area:
return self._vision_statements.get(area, {})
return self._vision_statements
async def check_alignment(
self,
proposed_action: str,
area: str = "general"
) -> Dict[str, Any]:
"""Check if a proposed action aligns with vision."""
vision = self._vision_statements.get(area, {})
if not vision:
await self.request_vision_clarification(area)
return {
"aligned": None,
"reason": "No vision statement recorded for this area",
"action": "Requested clarification from Kinan"
}
# This is a placeholder for more sophisticated alignment checking
# In a real implementation, this could use semantic similarity
return {
"aligned": True,
"vision": vision.get("statement"),
"priorities": vision.get("priorities"),
"confidence": 0.8
}
async def request_vision_clarification(
self,
area: str,
specific_questions: List[str] = None
) -> None:
"""Request clarification on vision for an area."""
questions = specific_questions or [
f"What is the strategic vision for {area}?",
f"What are the top priorities in {area}?",
f"What values should guide decisions in {area}?"
]
await self.channel.send(
subject=f"Vision Clarification Needed: {area}",
content=f"""
I need clarification on the vision for: {area}
This will help me make better aligned decisions.
Questions:
{chr(10).join(f' {i+1}. {q}' for i, q in enumerate(questions))}
Please share your thoughts when you have time.
""",
category=MessageCategory.VISION,
priority=MessagePriority.NORMAL,
requires_response=True
)
async def validate_priorities(
self,
current_priorities: List[str],
area: str = "general"
) -> None:
"""Validate current priorities against Kinan's vision."""
await self.channel.send(
subject=f"Priority Validation: {area}",
content=f"""
I want to validate my current priorities align with your vision.
Area: {area}
Current Priorities:
{chr(10).join(f' {i+1}. {p}' for i, p in enumerate(current_priorities))}
Please confirm these are correct or provide adjustments.
""",
category=MessageCategory.VISION,
priority=MessagePriority.LOW,
requires_response=True
)
async def report_alignment_score(
self,
area: str,
score: float,
factors: Dict[str, float]
) -> None:
"""Report current alignment score to Kinan."""
self._alignment_scores[area] = score
factors_text = "\n".join(
f" - {factor}: {value:.0%}"
for factor, value in factors.items()
)
await self.channel.send(
subject=f"Alignment Score: {area} - {score:.0%}",
content=f"""
Vision Alignment Report: {area}
Overall Score: {score:.0%}
Factor Breakdown:
{factors_text}
{"Areas for improvement identified." if score < 0.8 else "Strong alignment maintained."}
""",
category=MessageCategory.STATUS,
priority=MessagePriority.BACKGROUND
)
async def suggest_realignment(
self,
area: str,
current_state: str,
suggested_change: str,
rationale: str
) -> None:
"""Suggest a realignment to better match vision."""
await self.channel.send(
subject=f"Realignment Suggestion: {area}",
content=f"""
I've identified a potential alignment improvement.
Area: {area}
Current State:
{current_state}
Suggested Change:
{suggested_change}
Rationale:
{rationale}
Please advise if this aligns with your vision.
""",
category=MessageCategory.RECOMMENDATION,
priority=MessagePriority.NORMAL,
requires_response=True
)
class KinanLiaisonSystem:
"""
Complete Kinan Liaison System - integrating all communication components.
This is the main entry point for AIVA Queen to communicate with the founder.
"""
def __init__(
self,
db_path: str = "kinan_liaison.db",
slack_webhook: Optional[str] = None,
email_config: Optional[Dict[str, str]] = None,
default_channels: List[DeliveryChannel] = None
):
# Initialize persistence
self.persistence = MessagePersistence(db_path=db_path)
# Initialize delivery service
self.delivery = DeliveryService(
slack_webhook=slack_webhook,
email_config=email_config
)
# Initialize primary channel
self.channel = KinanChannel(
persistence=self.persistence,
delivery_service=self.delivery,
default_channels=default_channels or [DeliveryChannel.INTERNAL, DeliveryChannel.TERMINAL]
)
# Initialize component systems
self.resources = ResourceRequester(self.channel, self.persistence)
self.strategy = StrategyDiscussion(self.channel, self.persistence)
self.status = StatusReporter(self.channel)
self.approvals = ApprovalSeeker(self.channel, self.persistence)
self.vision = VisionAligner(self.channel, self.persistence)
logger.info("KinanLiaisonSystem initialized")
async def start(self):
"""Start all liaison system components."""
await self.channel.start()
await self.status.start_scheduler()
await self.approvals.start_auto_approve_checker()
logger.info("KinanLiaisonSystem fully started")
async def stop(self):
"""Stop all liaison system components."""
await self.channel.stop()
await self.status.stop_scheduler()
await self.approvals.stop()
logger.info("KinanLiaisonSystem stopped")
async def send_to_kinan(
self,
subject: str,
content: str,
**kwargs
) -> FounderMessage:
"""Convenience method to send a message to Kinan."""
return await self.channel.send(subject, content, **kwargs)
def get_pending_items(self) -> Dict[str, Any]:
"""Get all pending items requiring Kinan's attention."""
return {
"resource_requests": [r.to_dict() for r in self.resources.get_pending_requests()],
"approval_requests": [a.to_dict() for a in self.approvals.get_pending_approvals()],
"database": self.persistence.get_pending_requests()
}
def get_statistics(self) -> Dict[str, Any]:
"""Get comprehensive liaison statistics."""
return {
"persistence": self.persistence.get_statistics(),
"pending_resources": len(self.resources.get_pending_requests()),
"pending_approvals": len(self.approvals.get_pending_approvals()),
"vision_areas": list(self.vision._vision_statements.keys())
}
# Example usage and testing
async def example_usage():
"""Demonstrate the Kinan Liaison System."""
# Initialize the system (uses Elestio PostgreSQL)
liaison = KinanLiaisonSystem(
default_channels=[DeliveryChannel.TERMINAL, DeliveryChannel.INTERNAL]
)
await liaison.start()
# Send a status message
await liaison.send_to_kinan(
subject="Genesis System Online",
content="AIVA Queen has completed initialization. All systems nominal.",
category=MessageCategory.STATUS
)
# Request an API key
api_request = await liaison.resources.request_api_key(
service_name="OpenAI GPT-4",
justification="Required for advanced reasoning in Genesis Darwin Mode",
blocking_tasks=["Darwin competitive evaluation", "Axiom generation"]
)
print(f"API Request ID: {api_request.request_id}")
# Report a milestone
await liaison.status.report_milestone(
milestone_name="First Revenue Generated",
description="Successfully closed first Voice AI client for $2,500",
impact="Validates product-market fit and pricing strategy",
next_steps=["Scale outreach", "Document case study", "Refine onboarding"]
)
# Seek approval for a decision
approval = await liaison.approvals.seek_approval(
action_description="Upgrade Elestio infrastructure to next tier",
impact_assessment={
"cost_increase": "$50/month",
"performance_gain": "3x processing capacity",
"reliability": "99.99% uptime SLA"
},
risk_analysis={
"risk_level": "Low",
"rollback": "Can downgrade within 24 hours"
},
recommended_action="Approve upgrade for improved Genesis performance"
)
print(f"Approval Request ID: {approval.request_id}")
# Share a learning
await liaison.channel.share_learning(
insight="Database reactivation campaigns show 15% response rate with Australian slang",
source_context="A/B test across 500 leads in Florida plumbing vertical",
implications=[
"Localization significantly impacts engagement",
"Consider expanding Australian slang patterns",
"Test in other verticals"
],
confidence=0.85
)
# Check pending items
pending = liaison.get_pending_items()
print(f"\nPending items: {json.dumps(pending, indent=2, default=str)}")
# Get statistics
stats = liaison.get_statistics()
print(f"\nStatistics: {json.dumps(stats, indent=2)}")
await liaison.stop()
if __name__ == "__main__":
# Configure logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
# Run the example
asyncio.run(example_usage())