from __future__ import annotations import hashlib import json import time from app.ai.service import AiStudioService from app.copilot.actions import CopilotActionRegistry from app.copilot.errors import ( CopilotConfirmationRequiredError, CopilotInvalidRequestError, CopilotRunConflictError, ) from app.copilot.models import CopilotRunRecord from app.copilot.planner import CopilotPlanner from app.copilot.repository import CopilotRepository from app.copilot.schemas import ( CopilotCapabilities, CopilotContext, CopilotContextInput, CopilotEditorSummary, CopilotExecuteRequest, CopilotPlan, CopilotRun, CopilotRunCreate, CopilotRunList, ) from app.core.exceptions import MediaAPIError from app.core.logger import get_logger from app.projects.errors import ProjectEditorNotFoundError from app.projects.services.editor_service import ProjectEditorService from app.projects.services.project_service import ProjectService from app.security.assets import CanonicalAssetNotFoundError, CanonicalAssetService from app.security.audit import AuditService logger = get_logger(__name__) class CopilotService: def __init__( self, *, repository: CopilotRepository, planner: CopilotPlanner, actions: CopilotActionRegistry, projects: ProjectService, assets: CanonicalAssetService, editor: ProjectEditorService, ai: AiStudioService, audit: AuditService, ) -> None: self.repository = repository self.planner = planner self.actions = actions self.projects = projects self.assets = assets self.editor = editor self.ai = ai self.audit = audit async def capabilities( self, *, workspace_id: str, user_id: str, context: CopilotContextInput ) -> CopilotCapabilities: bounded = await self.build_context( workspace_id=workspace_id, user_id=user_id, supplied=context ) available = set(bounded.available_capabilities) return CopilotCapabilities( available=True, planner="deterministic", actions=self.actions.capabilities(available), permissions=["copilot:read", "copilot:execute"], ) async def build_context( self, *, workspace_id: str, user_id: str, supplied: CopilotContextInput, ) -> CopilotContext: project_id = str(supplied.project_id) if supplied.project_id else None capabilities = { "project.open", "asset.select", "template.search", "template.get", "template.apply", "template.create_project", "publishing.validate", "publishing.create_post", "publishing.schedule", "publishing.publish", "publishing.cancel", "analytics.overview", "analytics.sync", } editor_summary = None editor_document = None if project_id: await self.projects.get( workspace_id=workspace_id, user_id=user_id, project_id=project_id ) try: editor = await self.editor.get( workspace_id=workspace_id, user_id=user_id, project_id=project_id, ) editor_document = editor.state editor_summary = CopilotEditorSummary( revision=editor.revision, duration_ms=editor.state.duration_ms(), track_count=len(editor.state.timeline.tracks), clip_count=sum(len(track.clips) for track in editor.state.timeline.tracks), ) capabilities.update( { "editor.split_clip", "editor.delete_clip", "editor.set_duration", "editor.add_clip", "editor.render", } ) except ProjectEditorNotFoundError: editor_summary = None selected_assets = [] for asset_id in supplied.selected_asset_ids: try: asset = await self.assets.get_owned_by_id( workspace_id=workspace_id, user_id=user_id, asset_id=str(asset_id), ) except CanonicalAssetNotFoundError as exc: raise CopilotInvalidRequestError( "A selected asset is not available in this workspace." ) from exc if project_id and asset.project_id != project_id: raise CopilotInvalidRequestError( "Every selected asset must belong to the selected project." ) selected_assets.append(asset.id) selected_clips = list(dict.fromkeys(supplied.selected_clip_ids)) if selected_clips: if editor_document is None: raise CopilotInvalidRequestError("Selected clips require saved editor state.") known = {clip.id for track in editor_document.timeline.tracks for clip in track.clips} if any(clip_id not in known for clip_id in selected_clips): raise CopilotInvalidRequestError( "A selected clip is not present in the authoritative editor state." ) ai_capabilities = self.ai.capabilities() for tool in ai_capabilities.tools: if tool.available: capabilities.add(f"ai.{tool.operation.removeprefix('generate_')}") capabilities.add(f"ai.{tool.operation}") return CopilotContext( workspace_id=workspace_id, project_id=supplied.project_id, selected_asset_ids=selected_assets, selected_clip_ids=selected_clips, active_tool=supplied.active_tool, editor_summary=editor_summary, available_capabilities=sorted(capabilities), ) async def create_run( self, *, workspace_id: str, user_id: str, api_key_id: str, request_id: str, payload: CopilotRunCreate, idempotency_key: str, ) -> CopilotRun: key = idempotency_key.strip() if not key or len(key) > 255: raise CopilotInvalidRequestError("A bounded Idempotency-Key is required.") context = await self.build_context( workspace_id=workspace_id, user_id=user_id, supplied=payload.context ) fingerprint = hashlib.sha256( json.dumps( { "request": payload.request, "context": context.model_dump(mode="json"), }, sort_keys=True, separators=(",", ":"), ).encode() ).hexdigest() plan = self.planner.plan(payload.request, context) self.actions.validate_plan(plan.actions) record = CopilotRunRecord( workspace_id=workspace_id, user_id=user_id, project_id=str(context.project_id) if context.project_id else None, idempotency_key=key, request_fingerprint=fingerprint, request_text=payload.request, context_json=context.model_dump(mode="json"), plan_json=plan.model_dump(mode="json"), status="plan_ready" if plan.executable else "blocked", results_json=[], ) created, is_new = await self.repository.create(record) if is_new: await self.audit.record_event( workspace_id=workspace_id, user_id=user_id, api_key_id=api_key_id, request_id=request_id, event_type="copilot.run_started", entity_type="copilot_run", entity_id=created.id, metadata={"project_id": created.project_id}, ) await self.audit.record_event( workspace_id=workspace_id, user_id=user_id, api_key_id=api_key_id, request_id=request_id, event_type="copilot.plan_created", entity_type="copilot_run", entity_id=created.id, metadata={ "action_count": len(plan.actions), "requires_confirmation": plan.requires_confirmation, "executable": plan.executable, }, ) logger.info( "copilot plan created", extra={ "run_id": created.id, "project_id": created.project_id, "status": created.status, "action_count": len(plan.actions), }, ) return self._response(created) async def execute( self, *, workspace_id: str, user_id: str, api_key_id: str, request_id: str, run_id: str, payload: CopilotExecuteRequest, permissions: frozenset[str], ) -> CopilotRun: record = await self.repository.get(workspace_id, user_id, run_id) if record.status not in {"plan_ready"}: raise CopilotRunConflictError("Only a plan-ready Copilot run can be executed.") plan = CopilotPlan.model_validate(record.plan_json) self.actions.validate_plan(plan.actions) if not plan.executable: raise CopilotRunConflictError("This Copilot plan is not executable.") if plan.requires_confirmation and not payload.confirmed: raise CopilotConfirmationRequiredError( "Explicit confirmation is required before this plan can run." ) record = await self.repository.claim_execution( workspace_id, user_id, run_id, confirmed=payload.confirmed, ) results = [] started = time.monotonic() available = set(CopilotContext.model_validate(record.context_json).available_capabilities) for action in plan.actions: await self.repository.update( workspace_id, user_id, run_id, status="executing", current_action_id=action.id, results=[item.model_dump(mode="json") for item in results], ) await self.audit.record_event( workspace_id=workspace_id, user_id=user_id, api_key_id=api_key_id, request_id=request_id, event_type="copilot.action_started", entity_type="copilot_run", entity_id=run_id, metadata={"action_id": action.id, "action_type": action.type}, ) try: result = await self.actions.execute( action, workspace_id=workspace_id, user_id=user_id, api_key_id=api_key_id, request_id=request_id, run_id=run_id, permissions=permissions, available_capabilities=available, ) results.append(result) await self.audit.record_event( workspace_id=workspace_id, user_id=user_id, api_key_id=api_key_id, request_id=request_id, event_type="copilot.action_completed", entity_type="copilot_run", entity_id=run_id, metadata={ "action_id": action.id, "action_type": action.type, "resource_type": result.resource_type, "resource_id": result.resource_id, }, ) except Exception as exc: if isinstance(exc, MediaAPIError): error_code = exc.code error_message = exc.message else: error_code = "COPILOT_ACTION_FAILED" error_message = "The Copilot action failed unexpectedly." logger.error( "copilot action failed unexpectedly", extra={ "run_id": run_id, "project_id": record.project_id, "action_id": action.id, "action_type": action.type, "error_category": error_code, }, ) results.append( self._failed_result( action.id, action.type, error_message, error_code, ) ) await self.audit.record_event( workspace_id=workspace_id, user_id=user_id, api_key_id=api_key_id, request_id=request_id, event_type="copilot.action_failed", entity_type="copilot_run", entity_id=run_id, metadata={ "action_id": action.id, "action_type": action.type, "error_code": error_code, }, ) break completed = sum(item.status == "completed" for item in results) failed = sum(item.status == "failed" for item in results) if failed and completed: status = "partial" summary = ( f"{completed} of {len(plan.actions)} actions completed; " "the remaining workflow stopped after a failure." ) event = "copilot.run_failed" elif failed: status = "failed" summary = "The Copilot action failed before the workflow completed." event = "copilot.run_failed" else: status = "completed" summary = f"{completed} action{'s' if completed != 1 else ''} completed." event = "copilot.run_completed" updated = await self.repository.update( workspace_id, user_id, run_id, status=status, current_action_id=None, results=[item.model_dump(mode="json") for item in results], summary=summary, error_code=results[-1].error_code if failed else None, error_message=results[-1].summary if failed else None, terminal=True, ) await self.audit.record_event( workspace_id=workspace_id, user_id=user_id, api_key_id=api_key_id, request_id=request_id, event_type=event, entity_type="copilot_run", entity_id=run_id, metadata={ "status": status, "completed_actions": completed, "failed_actions": failed, }, ) logger.info( "copilot run finished", extra={ "run_id": run_id, "project_id": record.project_id, "status": status, "duration_ms": round((time.monotonic() - started) * 1_000), }, ) return self._response(updated) async def cancel( self, *, workspace_id: str, user_id: str, api_key_id: str, request_id: str, run_id: str, ) -> CopilotRun: updated, changed = await self.repository.cancel_before_execution( workspace_id, user_id, run_id, ) if not changed: return self._response(updated) await self.audit.record_event( workspace_id=workspace_id, user_id=user_id, api_key_id=api_key_id, request_id=request_id, event_type="copilot.run_cancelled", entity_type="copilot_run", entity_id=run_id, metadata={"project_id": updated.project_id}, ) return self._response(updated) async def get(self, *, workspace_id: str, user_id: str, run_id: str) -> CopilotRun: return self._response(await self.repository.get(workspace_id, user_id, run_id)) async def list( self, *, workspace_id: str, user_id: str, offset: int, limit: int, ) -> CopilotRunList: records = await self.repository.list(workspace_id, user_id, offset=offset, limit=limit) return CopilotRunList( items=[self._response(record) for record in records], offset=offset, limit=limit, ) @staticmethod def _failed_result(action_id: str, action_type: str, message: str, error_code: str): from app.copilot.schemas import CopilotActionResult return CopilotActionResult( action_id=action_id, action_type=action_type, status="failed", summary=message[:1_000], error_code=error_code, retryable=isinstance(error_code, str) and error_code in { "GENERATION_PROVIDER_UNAVAILABLE", "RATE_LIMIT_EXCEEDED", "PROJECT_EDITOR_REVISION_CONFLICT", }, ) @staticmethod def _response(record: CopilotRunRecord) -> CopilotRun: return CopilotRun( id=record.id, project_id=record.project_id, status=record.status, request=record.request_text, context=CopilotContext.model_validate(record.context_json), plan=CopilotPlan.model_validate(record.plan_json), current_action_id=record.current_action_id, results=record.results_json or [], summary=record.summary, error_code=record.error_code, error_message=record.error_message, created_at=record.created_at, updated_at=record.updated_at, completed_at=record.completed_at, )