103 lines
3.7 KiB
Python
103 lines
3.7 KiB
Python
"""
|
|
Phase 1 #1A: Skill Runtime Framework
|
|
Core executor architecture for real skill execution.
|
|
"""
|
|
# SkillRuntime class - lives in new file skill_runtime.py
|
|
class SkillRuntime:
|
|
"""Real skill execution engine.
|
|
|
|
Lifecycle:
|
|
authorize() -> prepare_workspace() -> execute() -> store_artifacts() -> record_metrics()
|
|
|
|
States: pending -> running -> ready | failed
|
|
"""
|
|
|
|
def __init__(self, skill_id: str, user_id: str, tenant_id: str = None):
|
|
self.skill_id = skill_id
|
|
self.user_id = user_id
|
|
self.tenant_id = tenant_id or "00000000-0000-4000-a000-000000000001"
|
|
self.instance_id = None
|
|
self.workspace = f"/workspace/{self.tenant_id}/{self.user_id}/{skill_id}"
|
|
self.status = "pending"
|
|
self.result = None
|
|
self.error = None
|
|
|
|
async def authorize(self, db) -> bool:
|
|
"""Check user has permission to execute this skill."""
|
|
from sqlalchemy import text
|
|
row = db.execute(text(
|
|
"SELECT 1 FROM app_installations WHERE user_id = :uid AND app_id = :sid AND status = 'active'"
|
|
), {"uid": self.user_id, "sid": self.skill_id}).fetchone()
|
|
return row is not None
|
|
|
|
async def prepare_workspace(self):
|
|
"""Create isolated workspace directory."""
|
|
import os
|
|
os.makedirs(self.workspace, exist_ok=True)
|
|
return self.workspace
|
|
|
|
async def execute(self, inputs: dict) -> dict:
|
|
"""Execute the skill with provided inputs."""
|
|
self.status = "running"
|
|
try:
|
|
# Subclass or adapter implements actual execution
|
|
self.result = await self._run(inputs)
|
|
self.status = "ready"
|
|
except Exception as e:
|
|
self.status = "failed"
|
|
self.error = str(e)
|
|
return {"status": self.status, "result": self.result, "error": self.error}
|
|
|
|
async def _run(self, inputs: dict) -> dict:
|
|
"""Override in skill-specific subclasses."""
|
|
raise NotImplementedError("Each skill must implement _run()")
|
|
|
|
async def store_artifacts(self):
|
|
"""Save generated files to workspace."""
|
|
import os, json
|
|
artifact_path = os.path.join(self.workspace, "result.json")
|
|
with open(artifact_path, "w") as f:
|
|
json.dump({"status": self.status, "result": self.result, "error": self.error}, f)
|
|
return artifact_path
|
|
|
|
async def record_metrics(self):
|
|
"""Log execution to DB."""
|
|
from app.database import SessionLocal
|
|
from sqlalchemy import text
|
|
db = SessionLocal()
|
|
db.execute(text(
|
|
"INSERT INTO skill_instances (id, skill_id, user_id, tenant_id, status, result, created_at) "
|
|
"VALUES (gen_random_uuid(), :sid, :uid, :tid, :st, :res, NOW())"
|
|
), {"sid": self.skill_id, "uid": self.user_id, "tid": self.tenant_id,
|
|
"st": self.status, "res": str(self.result)[:500] if self.result else None})
|
|
db.commit()
|
|
db.close()
|
|
|
|
|
|
class PPTSkillRuntime(SkillRuntime):
|
|
"""Skill-specific runtime for AI PPT Generator."""
|
|
|
|
async def _run(self, inputs: dict) -> dict:
|
|
content = inputs.get("content", "")
|
|
template = inputs.get("template", "business")
|
|
# Placeholder: calls external PPT generation API
|
|
return {
|
|
"slides": 5,
|
|
"title": f"PPT: {content[:50]}",
|
|
"template": template,
|
|
"status": "generated"
|
|
}
|
|
|
|
|
|
class SEOSkillRuntime(SkillRuntime):
|
|
"""Skill-specific runtime for SEO Optimizer."""
|
|
|
|
async def _run(self, inputs: dict) -> dict:
|
|
url = inputs.get("url", "")
|
|
keywords = inputs.get("keywords", [])
|
|
return {
|
|
"score": 85,
|
|
"suggestions": ["Add meta description", "Improve H1 tags"],
|
|
"url": url
|
|
}
|