from __future__ import annotations import asyncio import os import shutil from datetime import datetime, timezone from pathlib import Path from .models import AgentSession, ControlError, Project from .projects import ProjectRegistry from .rpc import JsonProcess from .settings import idle_only, validate_settings class PiAdapter: agent_type = "pi" def __init__(self, projects: ProjectRegistry, live_registry, rpc=None, approvals=None, user_inputs=None): self.projects, self.live_registry = projects, live_registry self.approvals = approvals self.user_inputs=user_inputs;self.native_tasks={};self.created_callbacks={} self.lock = asyncio.Lock() repo = Path(__file__).resolve().parents[3] node = os.environ.get("GATEWAY_NODE_BIN") or shutil.which("node") or "node" pi_binary = shutil.which("pi") sdk = os.environ.get("PI_SDK_MODULE", "") if not sdk and pi_binary: sdk = str(Path(pi_binary).resolve().parents[2] / "dist/index.js") env = {**os.environ, "PI_SDK_MODULE": sdk, "DEVICE_LAB_SDK_HOST": "1"} env.setdefault("DEVICE_LAB_CREDENTIALS_FILE", os.environ.get("DEVICE_CREDENTIALS_FILE", str(Path.home() / ".config/esp32-device-gateway/device.json"))) env.setdefault("DEVICE_LAB_GATEWAY_URL", f"http://127.0.0.1:{os.environ.get('GATEWAY_PORT', '8000')}") self.rpc = rpc or JsonProcess([node,"--experimental-strip-types","--experimental-import-meta-resolve",str(repo / "integrations/pi/sdk-host.mjs")], env=env,on_event=self.event,on_disconnect=self.disconnected) def disconnected(self): for task in self.native_tasks.values():task.cancel() self.native_tasks.clear() if self.user_inputs:self.user_inputs.cancel_owner('pi-sdk') async def event(self,message): params=message.get('params',{}) if message['method']=='session/created': callback=self.created_callbacks.get(params.get('operation_id')) if callback:callback(params['session_id']) elif message['method']=='native/closed': task=self.native_tasks.pop(params['id'],None) if task:task.cancel() elif message['method']=='native/userInput' and self.user_inputs: task=asyncio.create_task(self.native_input(params,self.rpc.generation)) self.native_tasks[params['id']]=task async def native_input(self,params,generation): entry=None try: project=next((p for p in self.projects.list() if p.path==params['cwd']),None) if not project:raise ControlError('Unregistered Pi cwd') entry=self.user_inputs.create('pi',params['session_id'],project.id,params['id'],params['questions'],'pi-sdk') result=await self.user_inputs.wait(entry) if self.rpc.generation==generation:await self.rpc.write({'type':'native_user_input_response','id':params['id'],'result':result}) except asyncio.CancelledError:pass finally: if entry:self.user_inputs.cancel(entry) async def creation_models(self,project): return await self.rpc.call('models',{'cwd':project.path}) async def read_history(self,session_id,max_messages=64): """Read-only persisted user/assistant text. Never resumes the session. The Pi SDK returns the session tail oldest -> newest. The unified history layer is newest-first, so the adapter normalizes the direction here; the SDK's own ordering never leaks into pagination. """ from .history import normalize_pi_messages # Resolve the canonical cwd through the existing registered metadata # rather than guessing a path; the SDK host only opens local sessions. session=await self.get_session(session_id) result=await self.rpc.call('history',{'session_id':session_id,'cwd':session.cwd,'max_messages':max_messages}) messages=result.get('messages') if isinstance(result,dict) else None return list(reversed(normalize_pi_messages(messages))) async def create_session(self,project,text,settings,on_created,operation_id=None): if settings.permissions is not None:raise ControlError('Pi permissions are native and read-only',400) models=await self.creation_models(project) validate_settings(settings,{'model':None,'effort':None,'available_models':models}) async with self.lock: self.created_callbacks[operation_id]=on_created try: result=await self.rpc.call('create',dict(cwd=project.path,text=text,model=settings.model,effort=settings.effort,operation_id=operation_id)) on_created(result['session_id']);return dict(result,project_id=project.id) finally:self.created_callbacks.pop(operation_id,None) def live_hint(self, session_id: str): item = self.live_registry.get(session_id) if item and item.state != "OFFLINE" and (datetime.now(timezone.utc) - item.last_activity).total_seconds() < 45: return item return None async def list_sessions(self, project: Project) -> list[AgentSession]: if not Path(project.path).is_dir(): raise ControlError("registered project directory no longer exists", 409) catalog = await self.rpc.call("list", {"cwd": project.path}) result = [] for item in catalog: session = AgentSession(agent="pi", project_id=project.id, **item) overlay = self.live_registry.get(session.session_id) if overlay and (session.live or self.live_hint(session.session_id)): session.live = True session.state = overlay.state if overlay.state != "OFFLINE" else item["state"] session.updated_at = max(session.updated_at, overlay.last_activity.isoformat()) session.latest_activity = overlay.latest_activity or session.latest_activity session.recent_text = overlay.public()["recent_text"] or session.recent_text session.final_text = overlay.public()["final_text"] or ('' if overlay.state=='RUNNING' else session.final_text) session.activity_type = overlay.activity_type session.waiting_reason = overlay.waiting_reason result.append(session) # Live sessions can precede their first persisted assistant response. known = {s.session_id for s in result} for overlay in list(self.live_registry.sessions.values()): if overlay.session_id not in known and overlay.cwd == project.path and self.live_hint(overlay.session_id): result.append(AgentSession(agent="pi", project_id=project.id, session_id=overlay.session_id, name=overlay.name, cwd=overlay.cwd, state=overlay.state, live=True, updated_at=overlay.last_activity.isoformat(), latest_activity=overlay.latest_activity, final_text=overlay.public()["final_text"])) if self.user_inputs:result=[self.user_inputs.overlay(s) for s in result] if self.approvals: result = [self.approvals.overlay(s) for s in result] return sorted(result, key=lambda s: s.updated_at, reverse=True) async def get_session(self, session_id: str) -> AgentSession: for project in self.projects.list(): for session in await self.list_sessions(project): if session.session_id == session_id: return session raise ControlError("session not found in registered projects", 404) async def resume_session(self, session_id: str) -> bool: async with self.lock: session = await self.get_session(session_id) result = await self.rpc.call("resume", {"session_id": session_id, "cwd": session.cwd, "allow_resume": self.live_hint(session_id) is None}) return result["resumed"] async def send_input(self, session_id: str, text: str) -> dict: async with self.lock: session = await self.get_session(session_id) return await self.rpc.call("input", {"session_id": session_id, "cwd": session.cwd, "text": text, "allow_resume": self.live_hint(session_id) is None}) async def get_settings(self, session_id): session = await self.get_session(session_id) return await self.rpc.call("settings", {"session_id": session_id, "cwd": session.cwd, "allow_resume": self.live_hint(session_id) is None}) async def list_models(self, session_id=None): return (await self.get_settings(session_id))["available_models"] async def update_settings(self, session_id, value, *, still_valid=lambda: True): async with self.lock: if value.permissions is not None:raise ControlError("Pi permissions are native and read-only",400) session = await self.get_session(session_id) idle_only(session) current = await self.get_settings(session_id) model, effort = validate_settings(value, current) if not still_valid(): raise ControlError("settings request cancelled", 409) idle_only(await self.get_session(session_id)) return await self.rpc.call("update_settings", {"session_id": session_id, "cwd": session.cwd, "model":model,"effort":effort,"allow_resume": self.live_hint(session_id) is None}) async def close(self): self.disconnected() await self.rpc.close()