LifeOS.py

# lifeos_agent_prototype_fixed.py
# LifeOS - Multi-Agent Prototype (fixed)

import asyncio
import datetime
import json
import uuid

def now_ts() -> str:
    return datetime.datetime.utcnow().isoformat() + "Z"

class MemoryStore:
    def __init__(self):
        self.longterm = {}
        self.sessions = {}

    def set_fact(self, key, value):
        self.longterm[key] = value
        print(f"[Memory] set_fact {key} = {value}")

    def get_fact(self, key, default=None):
        return self.longterm.get(key, default)

    def append_session(self, session_id, entry):
        self.sessions.setdefault(session_id, []).append(entry)
        print(f"[Memory] session {session_id} append: {entry}")

    def get_session(self, session_id):
        return self.sessions.get(session_id, [])

class MCPTool:
    def __init__(self, name):
        self.name = name

    async def call(self, action, payload):
        # Simulated async call (replace with real API calls)
        await asyncio.sleep(0.05)
        print(f"[MCP:{self.name}] action={action} payload={json.dumps(payload)}")
        return {"status": "ok", "action": action, "result": payload}

# MCP tool stubs (replace with real connectors)
calendar_tool = MCPTool('calendar')
email_tool = MCPTool('email')
fitness_tool = MCPTool('fitness')
payment_tool = MCPTool('payment')

class Orchestrator:
    def __init__(self):
        self.agents = {}

    def register(self, agent):
        self.agents[agent.name] = agent
        print(f"[Orch] Registered agent: {agent.name}")

    async def route_message(self, msg):
        recipient = msg.get('to')
        if recipient == 'orchestrator':
            print(f"[Orch] internal message: {msg}")
            return
        agent = self.agents.get(recipient)
        if not agent:
            print(f"[Orch] unknown recipient {recipient}")
            return
        await agent.receive(msg)

    async def broadcast(self, sender, payload, to_list):
        for r in to_list:
            msg = {'from': sender, 'to': r, 'ts': now_ts(), 'id': str(uuid.uuid4()), 'payload': payload}
            await self.route_message(msg)

class Agent:
    def __init__(self, name, orchestrator, memory):
        self.name = name
        self.orch = orchestrator
        self.memory = memory
        self.inbox = asyncio.Queue()
        self.running = True

    async def start(self):
        print(f"[{self.name}] starting")
        while self.running:
            try:
                msg = await asyncio.wait_for(self.inbox.get(), timeout=1.0)
                await self.handle_message(msg)
            except asyncio.TimeoutError:
                await self.on_idle()

    async def stop(self):
        self.running = False

    async def send(self, recipient, payload):
        msg = {
            'from': self.name,
            'to': recipient,
            'ts': now_ts(),
            'id': str(uuid.uuid4()),
            'payload': payload,
        }
        await self.orch.route_message(msg)

    async def receive(self, msg):
        await self.inbox.put(msg)

    async def handle_message(self, msg):
        print(f"[{self.name}] received message: {msg}")

    async def on_idle(self):
        # Optional background work
        await asyncio.sleep(0)

class PlannerAgent(Agent):
    async def handle_message(self, msg):
        payload = msg.get('payload', {}) or {}
        if payload.get('type') == 'user_goal':
            goal = payload.get('goal', '') or ''
            session = payload.get('session')
            print(f"[Planner] planning for goal: {goal}")
            tasks = []
            g = goal.lower()
            if 'lose' in g or 'weight' in g or 'kg' in g:
                tasks.append({'task': 'create_fitness_plan', 'details': {'target': '5kg', 'by': '2026-03-01'}})
            if 'productivity' in g or 'productiv' in g:
                tasks.append({'task': 'optimize_schedule', 'details': {}})
            if 'save' in g or 'money' in g:
                tasks.append({'task': 'review_finances', 'details': {}})
            for t in tasks:
                if t['task'].startswith('create_fitness'):
                    await self.send('health_agent', {'type': 'task', 'task': t})
                elif t['task'] == 'optimize_schedule':
                    await self.send('work_agent', {'type': 'task', 'task': t})
                elif t['task'] == 'review_finances':
                    await self.send('finance_agent', {'type': 'task', 'task': t})

class HealthAgent(Agent):
    async def handle_message(self, msg):
        payload = msg.get('payload', {}) or {}
        if payload.get('type') == 'task':
            task = payload.get('task', {}) or {}
            if task.get('task') == 'create_fitness_plan':
                details = task.get('details', {}) or {}
                plan = self._generate_plan(details)
                self.memory.set_fact('fitness_plan', plan)
                await fitness_tool.call('create_plan', plan)
                # Notify work agent so schedule can be updated
                await self.send('work_agent', {'type': 'info', 'info': 'fitness_plan_created', 'plan': plan})

    def _generate_plan(self, details):
        return {
            'target': details.get('target', 'unknown'),
            'deadline': details.get('by', 'unknown'),
            'workouts_per_week': 4,
            'meals_per_day': 3,
        }

class WorkAgent(Agent):
    async def handle_message(self, msg):
        payload = msg.get('payload', {}) or {}
        if payload.get('type') == 'task':
            task = payload.get('task', {}) or {}
            if task.get('task') == 'optimize_schedule':
                await calendar_tool.call('fetch', {'range': 'next_30_days'})
                schedule = self._generate_schedule()
                self.memory.set_fact('work_schedule', schedule)
                await calendar_tool.call('upsert_events', {'events': schedule.get('events', [])})
                await self.send('comm_agent', {'type': 'info', 'info': 'schedule_updated', 'schedule': schedule})
        elif payload.get('type') == 'info' and payload.get('info') == 'fitness_plan_created':
            plan = payload.get('plan')
            print(f"[WorkAgent] adjusting schedule for fitness plan {plan}")

    def _generate_schedule(self):
        events = [
            {'title': 'Focus Block', 'start': '2025-11-18T09:00:00Z', 'duration_min': 90},
            {'title': 'Gym', 'start': '2025-11-18T18:00:00Z', 'duration_min': 60},
        ]
        return {'events': events}

class CommAgent(Agent):
    async def handle_message(self, msg):
        payload = msg.get('payload', {}) or {}
        if payload.get('type') == 'info' and payload.get('info') == 'schedule_updated':
            schedule = payload.get('schedule', {})
            digest = self._create_digest(schedule)
            await email_tool.call('send_summary', {'subject': "Today's schedule", 'body': digest})

    def _create_digest(self, schedule):
        return "Your schedule was updated. Events: " + json.dumps(schedule.get('events', []))

class FinanceAgent(Agent):
    async def handle_message(self, msg):
        payload = msg.get('payload', {}) or {}
        if payload.get('type') == 'task' and payload.get('task', {}).get('task') == 'review_finances':
            await payment_tool.call('fetch_recent', {})
            report = {'savings_opportunity': True, 'recommendation': 'Cut subscription X'}
            self.memory.set_fact('finance_report', report)
            await self.send('comm_agent', {'type': 'info', 'info': 'finance_report_ready', 'report': report})

class LearningAgent(Agent):
    async def handle_message(self, msg):
        payload = msg.get('payload', {}) or {}
        if payload.get('type') == 'user_goal' and 'learn' in (payload.get('goal', '') or ''):
            lessons = [{'title': 'Day 1 micro-lesson', 'content': '...'}]
            self.memory.set_fact('learning_path', lessons)
            await self.send('comm_agent', {'type': 'info', 'info': 'learning_path_ready', 'lessons': lessons})

async def user_input_flow(orchestrator, planner, memory):
    session_id = str(uuid.uuid4())
    user_goal = input('Enter a short life goal (e.g. "Lose 5kg before March and improve productivity"):\\n> ').strip()
    memory.append_session(session_id, {'user_goal': user_goal, 'ts': now_ts()})
    await planner.handle_message({'payload': {'type': 'user_goal', 'goal': user_goal, 'session': session_id}})

async def main():
    memory = MemoryStore()
    orch = Orchestrator()

    # instantiate agents
    planner = PlannerAgent('planner', orch, memory)
    health = HealthAgent('health_agent', orch, memory)
    work = WorkAgent('work_agent', orch, memory)
    comm = CommAgent('comm_agent', orch, memory)
    finance = FinanceAgent('finance_agent', orch, memory)
    learning = LearningAgent('learning_agent', orch, memory)

    # register agents
    for a in [planner, health, work, comm, finance, learning]:
        orch.register(a)

    # start agents (background)
    tasks = [asyncio.create_task(a.start()) for a in orch.agents.values()]

    print('\\nLifeOS prototype running.\\n')
    print('Options:')
    print('1) Interactive goal input')
    print('2) Run demo goal (Lose 5kg, improve productivity)')
    choice = input('Choose 1 or 2: ').strip()
    if choice == '1':
        await user_input_flow(orch, planner, memory)
    else:
        demo_goal = 'Lose 5 kg before March and improve my productivity'
        memory.append_session('demo-session', {'user_goal': demo_goal, 'ts': now_ts()})
        await planner.handle_message({'payload': {'type': 'user_goal', 'goal': demo_goal, 'session': 'demo-session'}})

    # give agents a moment to process
    await asyncio.sleep(1.0)

    print('\\n--- Memory snapshot ---')
    print(json.dumps({'longterm': memory.longterm, 'sessions': memory.sessions}, indent=2))

    # shutdown
    for a in orch.agents.values():
        await a.stop()
    await asyncio.gather(*tasks, return_exceptions=True)
    print('\\nLifeOS prototype stopped.')

if __name__ == '__main__':
    asyncio.run(main())