Spaces:
Running
Running
File size: 6,554 Bytes
dcbc7d2 01ab0a4 dcbc7d2 4d1c22f dcbc7d2 01ab0a4 dcbc7d2 4d1c22f dcbc7d2 01ab0a4 dcbc7d2 01ab0a4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 | # backend/tasks/executor.py
import asyncio
from dataclasses import dataclass, field
from typing import Any
from uuid import uuid4
@dataclass
class Task:
id: str = field(default_factory=lambda: str(uuid4()))
name: str = ""
steps: list[dict] = field(default_factory=list)
current_step: int = 0
status: str = "pending" # pending | running | paused | completed | failed | cancelled
result: Any = None
error: str = ""
permission_tier: str = "approval_required" # pre_approved | approval_required | explicit_each_time | biometric
class TaskExecutor:
def __init__(self, app_handle, permission_gateway):
self.tasks: dict[str, Task] = {}
self.app_handle = app_handle
self.permission_gateway = permission_gateway
# Load db path
import sys, os
if getattr(sys, 'frozen', False):
self.db_path = os.path.join(os.environ.get('APPDATA', os.path.expanduser('~')), 'JARVIS_OS', 'memory.db')
else:
self.db_path = os.path.join(os.getcwd(), 'memory.db')
self._recover_tasks()
def _recover_tasks(self):
import sqlite3
try:
with sqlite3.connect(self.db_path) as conn:
conn.execute('PRAGMA journal_mode=WAL')
cursor = conn.cursor()
cursor.execute("SELECT id, name, status, steps, current_step, result, error, permission_tier FROM tasks")
for row in cursor.fetchall():
status = row[2]
# Safety: Fail any tasks that were running/paused during a hard crash
if status in ["running", "paused"]:
status = "failed"
cursor.execute("UPDATE tasks SET status=?, error=? WHERE id=?",
("failed", "System rebooted mid-execution", row[0]))
conn.commit()
t = Task(
id=row[0], name=row[1], status=status,
steps=[], # Callables cannot be easily serialized to JSON, so we leave steps empty on recovery
current_step=row[4], result=row[5], error=row[6], permission_tier=row[7]
)
self.tasks[t.id] = t
except Exception as e:
import logging; logging.getLogger(__name__).error(f"Swallowed exception: {e}") # Tables might not exist yet
def _save_task(self, task: Task):
import sqlite3, json
try:
with sqlite3.connect(self.db_path) as conn:
conn.execute('PRAGMA journal_mode=WAL')
cursor = conn.cursor()
# We do not serialize the actual 'handler' since it's a function reference.
# We just serialize a dummy representation of steps for UI if needed.
steps_json = json.dumps([{"name": getattr(s.get('handler'), '__name__', 'step')} for s in task.steps])
cursor.execute('''
INSERT INTO tasks (id, name, status, steps, current_step, result, error, permission_tier)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET
status=excluded.status,
current_step=excluded.current_step,
result=excluded.result,
error=excluded.error
''', (task.id, task.name, task.status, steps_json, task.current_step, str(task.result) if task.result else "", task.error, task.permission_tier))
conn.commit()
except Exception as e:
import logging; logging.getLogger(__name__).error(f"Swallowed exception: {e}")
async def submit(self, task: Task) -> str:
self.tasks[task.id] = task
self._save_task(task)
asyncio.create_task(self._run(task))
return task.id
async def _run(self, task: Task):
task.status = "running"
self._save_task(task)
self._emit("task:updated", task)
for i, step in enumerate(task.steps):
task.current_step = i
self._save_task(task)
self._emit("task:step_started", {"task_id": task.id, "step": step})
allowed = await self.permission_gateway.check(task.permission_tier, step)
if not allowed:
task.status = "paused"
self._save_task(task)
self._emit("task:awaiting_permission", {"task_id": task.id, "step": step})
allowed = await self.permission_gateway.wait_for_user_response(task.id)
if not allowed:
task.status = "cancelled"
self._save_task(task)
self._emit("task:updated", task)
return
try:
result = await step["handler"](**step["params"])
task.result = result
self._save_task(task)
self._emit("task:step_completed", {"task_id": task.id, "step": step, "result": result})
except Exception as e:
task.status = "failed"
task.error = str(e)
self._save_task(task)
self._emit("task:updated", task)
return
task.status = "completed"
self._save_task(task)
self._emit("task:updated", task)
def cancel(self, task_id: str):
task = self.tasks.get(task_id)
if task:
task.status = "cancelled"
self._save_task(task)
self._emit("task:updated", task)
def _emit(self, event: str, payload):
try:
from backend.ws.agent_ws import ws_manager
import json
from dataclasses import asdict
# Serialize payload to dict safely
if hasattr(payload, '__dataclass_fields__'):
safe_payload = asdict(payload)
else:
# Fallback via json roundtrip to handle non-serializable objects cleanly
safe_payload = json.loads(json.dumps(payload, default=str))
asyncio.create_task(ws_manager.broadcast({"event": event, "payload": safe_payload}))
except Exception as e:
import logging; logging.getLogger(__name__).error(f"Swallowed exception: {e}")
|