from __future__ import annotations from dataclasses import dataclass from uuid import uuid4 from app.ai.schemas import AiGenerateImageRequest, AiGenerateVideoRequest from app.ai.service import AiStudioService from app.analytics.schemas import AnalyticsQuery, AnalyticsSyncRequest from app.analytics.service import AnalyticsDomainService from app.copilot.errors import ( CopilotCapabilityError, CopilotInvalidRequestError, CopilotPermissionError, ) from app.copilot.schemas import ( AiGenerateImageAction, AiGenerateVideoAction, AnalyticsOverviewAction, AnalyticsSyncAction, AssetSelectAction, CopilotAction, CopilotActionCapability, CopilotActionResult, EditorAddClipAction, EditorDeleteClipAction, EditorRenderAction, EditorSetDurationAction, EditorSplitClipAction, ProjectOpenAction, PublishingCancelAction, PublishingCreatePostAction, PublishingPublishAction, PublishingScheduleAction, PublishingValidateAction, TemplateApplyAction, TemplateCreateProjectAction, TemplateGetAction, TemplateSearchAction, ) from app.projects.editor_schemas import ( AudioClip, ClipTransform, EditorDocument, EditorSaveRequest, EffectClip, MediaClip, ProjectRenderCreate, SourceClip, Track, ) from app.projects.services.editor_service import ProjectEditorService from app.projects.services.project_service import ProjectService from app.projects.services.render_service import ProjectRenderService from app.security.assets import CanonicalAssetService from app.social.services.publishing_service import PublishingService from app.social.services.scheduling_service import SchedulingService from app.templates.marketplace_schemas import TemplateApply, TemplateInstantiate from app.templates.marketplace_service import MarketplaceTemplateService @dataclass(frozen=True, slots=True) class CopilotActionDefinition: type: str description: str required_permission: str required_capability: str destructive: bool external_side_effect: bool requires_confirmation: bool audit_event: str ACTION_DEFINITIONS = ( CopilotActionDefinition( "project.open", "Open a project.", "projects:read", "project.open", False, False, False, "copilot.project_opened", ), CopilotActionDefinition( "asset.select", "Open a canonical asset.", "assets:read", "asset.select", False, False, False, "copilot.asset_selected", ), CopilotActionDefinition( "ai.generate_image", "Submit image generation.", "ai:generate", "ai.generate_image", False, False, True, "copilot.ai_generation_requested", ), CopilotActionDefinition( "ai.generate_video", "Submit video generation.", "ai:generate", "ai.generate_video", False, False, True, "copilot.ai_generation_requested", ), CopilotActionDefinition( "editor.split_clip", "Split a timeline clip.", "projects:update", "editor.split_clip", False, False, False, "copilot.editor_updated", ), CopilotActionDefinition( "editor.delete_clip", "Delete a timeline clip.", "projects:update", "editor.delete_clip", True, False, True, "copilot.editor_updated", ), CopilotActionDefinition( "editor.set_duration", "Set a clip duration.", "projects:update", "editor.set_duration", False, False, False, "copilot.editor_updated", ), CopilotActionDefinition( "editor.add_clip", "Add an asset to the timeline.", "projects:update", "editor.add_clip", False, False, False, "copilot.editor_updated", ), CopilotActionDefinition( "editor.render", "Submit a project render.", "projects:update", "editor.render", False, False, True, "copilot.render_requested", ), CopilotActionDefinition( "template.search", "Search visible marketplace templates.", "templates:read", "template.search", False, False, False, "copilot.template_searched", ), CopilotActionDefinition( "template.get", "Inspect a visible marketplace template.", "templates:read", "template.get", False, False, False, "copilot.template_opened", ), CopilotActionDefinition( "template.apply", "Apply a template to an existing project.", "templates:apply", "template.apply", True, False, True, "copilot.template_applied", ), CopilotActionDefinition( "template.create_project", "Create a project from a template.", "templates:apply", "template.create_project", False, False, True, "copilot.template_project_created", ), CopilotActionDefinition( "publishing.validate", "Validate every social publishing target.", "social:posts:write", "publishing.validate", False, False, False, "copilot.publishing_validated", ), CopilotActionDefinition( "publishing.create_post", "Create a typed canonical social post draft.", "social:posts:write", "publishing.create_post", False, False, False, "copilot.publishing_draft_created", ), CopilotActionDefinition( "publishing.schedule", "Schedule an existing social post.", "social:schedules:write", "publishing.schedule", False, True, True, "copilot.publishing_scheduled", ), CopilotActionDefinition( "publishing.publish", "Publish an existing social post to its selected accounts.", "social:posts:publish", "publishing.publish", False, True, True, "copilot.publishing_started", ), CopilotActionDefinition( "publishing.cancel", "Cancel eligible publishing targets or request in-flight cancellation.", "social:posts:write", "publishing.cancel", False, True, True, "copilot.publishing_cancelled", ), CopilotActionDefinition( "analytics.overview", "Read authoritative analytics insights.", "analytics:read", "analytics.overview", False, False, False, "copilot.analytics_viewed", ), CopilotActionDefinition( "analytics.sync", "Queue authoritative analytics synchronization.", "analytics:sync", "analytics.sync", False, False, True, "copilot.analytics_sync_requested", ), ) class CopilotActionRegistry: def __init__( self, *, projects: ProjectService, assets: CanonicalAssetService, editor: ProjectEditorService, renders: ProjectRenderService, ai: AiStudioService, templates: MarketplaceTemplateService, publishing: PublishingService | None = None, scheduling: SchedulingService | None = None, analytics: AnalyticsDomainService | None = None, ) -> None: self.projects = projects self.assets = assets self.editor = editor self.renders = renders self.ai = ai self.templates = templates self.publishing = publishing self.scheduling = scheduling self.analytics = analytics self.definitions = {item.type: item for item in ACTION_DEFINITIONS} def validate(self, action: CopilotAction) -> CopilotActionDefinition: definition = self.definitions.get(action.type) if definition is None: raise CopilotInvalidRequestError("Copilot action type is not registered.") expected = { "required_permission": definition.required_permission, "required_capability": definition.required_capability, "destructive": definition.destructive, "external_side_effect": definition.external_side_effect, "requires_confirmation": definition.requires_confirmation, } if any(getattr(action, field) != value for field, value in expected.items()): raise CopilotInvalidRequestError( "Copilot action policy metadata does not match the registered action." ) return definition def validate_plan(self, actions: list[CopilotAction]) -> None: for action in actions: self.validate(action) def capabilities(self, available: set[str]) -> list[CopilotActionCapability]: return [ CopilotActionCapability( type=item.type, description=item.description, required_permission=item.required_permission, required_capability=item.required_capability, destructive=item.destructive, external_side_effect=item.external_side_effect, requires_confirmation=item.requires_confirmation, available=item.required_capability in available, ) for item in ACTION_DEFINITIONS ] async def execute( self, action: CopilotAction, *, workspace_id: str, user_id: str, api_key_id: str, request_id: str, run_id: str, permissions: frozenset[str], available_capabilities: set[str], ) -> CopilotActionResult: definition = self.validate(action) if definition.required_permission not in permissions and "admin" not in permissions: raise CopilotPermissionError( f"Permission '{definition.required_permission}' is required." ) if definition.required_capability not in available_capabilities: raise CopilotCapabilityError( f"Capability '{definition.required_capability}' is unavailable." ) if isinstance(action, EditorRenderAction) and not ( "jobs:create" in permissions or "admin" in permissions ): raise CopilotPermissionError("Permission 'jobs:create' is required.") if isinstance(action, TemplateApplyAction) and not ( "projects:update" in permissions or "admin" in permissions ): raise CopilotPermissionError("Permission 'projects:update' is required.") if isinstance(action, TemplateCreateProjectAction) and not ( "projects:create" in permissions or "admin" in permissions ): raise CopilotPermissionError("Permission 'projects:create' is required.") if ( isinstance(action, (TemplateApplyAction, TemplateCreateProjectAction)) and any( binding.asset_id is not None for binding in action.arguments.slot_bindings.values() ) and not ("assets:read" in permissions or "admin" in permissions) ): raise CopilotPermissionError("Permission 'assets:read' is required.") if isinstance( action, (AiGenerateVideoAction, EditorAddClipAction), ) and not ("assets:read" in permissions or "admin" in permissions): raise CopilotPermissionError("Permission 'assets:read' is required.") if ( isinstance(action, AiGenerateImageAction) and action.arguments.source_asset_id is not None and not ("assets:read" in permissions or "admin" in permissions) ): raise CopilotPermissionError("Permission 'assets:read' is required.") if isinstance(action, ProjectOpenAction): project = await self.projects.get( workspace_id=workspace_id, user_id=user_id, project_id=str(action.arguments.project_id), ) return self._success(action, "Project is ready to open.", "project", project.id) if isinstance(action, AssetSelectAction): asset = await self.assets.get_owned_by_id( workspace_id=workspace_id, user_id=user_id, asset_id=str(action.arguments.asset_id), ) if action.arguments.project_id is not None and asset.project_id != str( action.arguments.project_id ): raise CopilotInvalidRequestError( "The selected asset does not belong to the selected project." ) return self._success(action, "Asset is ready to open.", "asset", asset.id) if isinstance( action, ( PublishingValidateAction, PublishingCreatePostAction, PublishingScheduleAction, PublishingPublishAction, PublishingCancelAction, ), ): if self.publishing is None or self.scheduling is None: raise CopilotCapabilityError("Publishing orchestration is unavailable.") if isinstance(action, PublishingValidateAction): result = await self.publishing.validate_post_targets( workspace_id, str(action.arguments.post_id) ) return self._success( action, ( "Publishing targets are valid." if result.valid else "Publishing validation found issues." ), "social_post", result.post_id, ) if isinstance(action, PublishingCreatePostAction): result = await self.publishing.create( workspace_id=workspace_id, user_id=user_id, payload=action.arguments.post, idempotency_key=f"copilot:{run_id}:{action.id}", ) return self._success( action, "Publishing draft was created.", "social_post", result.id ) if isinstance(action, PublishingScheduleAction): validation = await self.publishing.validate_post_targets( workspace_id, str(action.arguments.post_id) ) if not validation.valid: raise CopilotInvalidRequestError( "Publishing validation must pass before scheduling." ) result = await self.scheduling.schedule( workspace_id, str(action.arguments.post_id), action.arguments.schedule ) return self._success( action, "Publishing was scheduled.", "social_schedule", result.id ) if isinstance(action, PublishingPublishAction): jobs = await self.publishing.queue( workspace_id, str(action.arguments.post_id), idempotency_key=f"copilot:{run_id}:{action.id}", ) return self._success( action, f"Queued {len(jobs)} publishing targets.", "social_post", str(action.arguments.post_id), ) result = await self.publishing.cancel(workspace_id, str(action.arguments.post_id)) return self._success( action, "Cancellation was applied to eligible publishing targets.", "social_post", result.id, ) if isinstance(action, (AnalyticsOverviewAction, AnalyticsSyncAction)): if self.analytics is None: raise CopilotCapabilityError("Analytics orchestration is unavailable.") if isinstance(action, AnalyticsOverviewAction): result = await self.analytics.overview( workspace_id, AnalyticsQuery( project_id=( str(action.arguments.project_id) if action.arguments.project_id else None ), provider=action.arguments.provider, metric=action.arguments.metric, timezone=action.arguments.timezone, sort=action.arguments.metric, ), ) freshness = result.freshness.status return self._success( action, f"Analytics overview is {freshness}; unavailable metrics were not inferred.", "analytics_overview", ( str(action.arguments.project_id) if action.arguments.project_id else workspace_id ), ) result = await self.analytics.create_sync( workspace_id, user_id, AnalyticsSyncRequest( project_id=( str(action.arguments.project_id) if action.arguments.project_id else None ), provider=action.arguments.provider, timezone=action.arguments.timezone, idempotency_key=f"copilot:{run_id}:{action.id}", ), ) return self._success( action, "Analytics synchronization was queued.", "analytics_sync", result.id ) if isinstance(action, TemplateSearchAction): result = await self.templates.list( workspace_id=workspace_id, user_id=user_id, search=action.arguments.query, category=action.arguments.category, aspect_ratio=None, min_duration_ms=None, max_duration_ms=None, media_type=None, visibility=None, status=None, capability=None, available_only=False, offset=0, limit=24, ) return self._success( action, f"Found {result.total} visible templates.", "template_search", action.arguments.query, ) if isinstance(action, TemplateGetAction): template = await self.templates.get( workspace_id=workspace_id, user_id=user_id, template_id=str(action.arguments.template_id), ) return self._success(action, "Template is ready to open.", "template", template.id) if isinstance(action, TemplateApplyAction): result = await self.templates.apply( workspace_id=workspace_id, user_id=user_id, api_key_id=api_key_id, request_id=request_id, template_id=str(action.arguments.template_id), payload=TemplateApply( project_id=action.arguments.project_id, template_version_id=action.arguments.template_version_id, slot_bindings=action.arguments.slot_bindings, ), idempotency_key=f"copilot:{run_id}:{action.id}", instantiate=False, ) return self._success( action, "Template was applied to the project.", "project", result.project_id ) if isinstance(action, TemplateCreateProjectAction): result = await self.templates.apply( workspace_id=workspace_id, user_id=user_id, api_key_id=api_key_id, request_id=request_id, template_id=str(action.arguments.template_id), payload=TemplateInstantiate( project_name=action.arguments.project_name, template_version_id=action.arguments.template_version_id, slot_bindings=action.arguments.slot_bindings, ), idempotency_key=f"copilot:{run_id}:{action.id}", instantiate=True, ) return self._success( action, "Project was created from the template.", "project", result.project_id ) if isinstance(action, AiGenerateImageAction): job = await self.ai.create( workspace_id=workspace_id, user_id=user_id, api_key_id=api_key_id, request_id=request_id, idempotency_key=f"copilot:{run_id}:{action.id}", payload=AiGenerateImageRequest( operation="generate_image", prompt=action.arguments.prompt, model=action.arguments.model, project_id=action.arguments.project_id, source_asset_ids=( [action.arguments.source_asset_id] if action.arguments.source_asset_id else [] ), ), ) return self._success( action, "Image generation was submitted.", "ai_generation", job.generation_id ) if isinstance(action, AiGenerateVideoAction): job = await self.ai.create( workspace_id=workspace_id, user_id=user_id, api_key_id=api_key_id, request_id=request_id, idempotency_key=f"copilot:{run_id}:{action.id}", payload=AiGenerateVideoRequest( operation="generate_video", prompt=action.arguments.prompt, model=action.arguments.model, project_id=action.arguments.project_id, source_asset_ids=[action.arguments.source_asset_id], ), ) return self._success( action, "Video generation was submitted.", "ai_generation", job.generation_id ) if isinstance(action, EditorRenderAction): render = await self.renders.create( workspace_id=workspace_id, user_id=user_id, api_key_id=api_key_id, request_id=request_id, project_id=str(action.arguments.project_id), idempotency_key=f"copilot:{run_id}:{action.id}", payload=ProjectRenderCreate(editor_revision=action.arguments.expected_revision), ) return self._success(action, "Render was submitted.", "project_render", render.id) if isinstance( action, ( EditorSplitClipAction, EditorDeleteClipAction, EditorSetDurationAction, EditorAddClipAction, ), ): revision = await self._execute_editor_action( action, workspace_id=workspace_id, user_id=user_id, api_key_id=api_key_id, request_id=request_id, ) return self._success( action, f"Editor revision {revision} was saved.", "project_editor_revision", str(revision), ) raise CopilotInvalidRequestError("Copilot action type is not registered.") async def _execute_editor_action( self, action, *, workspace_id: str, user_id: str, api_key_id: str, request_id: str, ) -> int: project_id = str(action.arguments.project_id) current = await self.editor.get( workspace_id=workspace_id, user_id=user_id, project_id=project_id ) if current.revision != action.arguments.expected_revision: raise CopilotInvalidRequestError( "The editor changed after this plan was created. Create a new plan." ) document = current.state.model_copy(deep=True) if isinstance(action, EditorSplitClipAction): self._split(document, action.arguments.clip_id, action.arguments.at_ms) elif isinstance(action, EditorDeleteClipAction): self._delete(document, action.arguments.clip_id) elif isinstance(action, EditorSetDurationAction): self._set_duration(document, action.arguments.clip_id, action.arguments.duration_ms) elif isinstance(action, EditorAddClipAction): await self._add_clip( document, workspace_id=workspace_id, user_id=user_id, project_id=project_id, asset_id=str(action.arguments.asset_id), duration_ms=action.arguments.duration_ms, ) validated = EditorDocument.model_validate(document.model_dump(by_alias=True)) saved = await self.editor.save( workspace_id=workspace_id, user_id=user_id, api_key_id=api_key_id, request_id=request_id, project_id=project_id, payload=EditorSaveRequest( expected_revision=current.revision, schema_version=validated.schema_version, state=validated, ), ) return saved.revision @staticmethod def _find_clip(document: EditorDocument, clip_id: str): for track in document.timeline.tracks: for index, clip in enumerate(track.clips): if clip.id == clip_id: return track, index, clip raise CopilotInvalidRequestError("The selected clip no longer exists.") def _split(self, document: EditorDocument, clip_id: str, at_ms: int) -> None: track, index, clip = self._find_clip(document, clip_id) relative = at_ms - clip.start_ms if relative <= 0 or relative >= clip.duration_ms: raise CopilotInvalidRequestError("The split point must fall inside the selected clip.") right = clip.model_copy(deep=True) right.id = str(uuid4()) right.start_ms = at_ms right.duration_ms = clip.duration_ms - relative clip.duration_ms = relative if isinstance(clip, SourceClip) and isinstance(right, SourceClip): right.source_start_ms = clip.source_start_ms + relative right.source_duration_ms = right.duration_ms clip.source_duration_ms = clip.duration_ms track.clips.insert(index + 1, right) def _delete(self, document: EditorDocument, clip_id: str) -> None: track, index, _ = self._find_clip(document, clip_id) track.clips.pop(index) document.timeline.transitions = [ item for item in document.timeline.transitions if item.from_clip_id != clip_id and item.to_clip_id != clip_id ] for candidate in document.timeline.tracks: candidate.clips = [ clip for clip in candidate.clips if not (isinstance(clip, EffectClip) and clip.target_clip_id == clip_id) ] def _set_duration(self, document: EditorDocument, clip_id: str, duration_ms: int) -> None: _, _, clip = self._find_clip(document, clip_id) if isinstance(clip, SourceClip) and not ( isinstance(clip, MediaClip) and clip.media_type == "image" ): raise CopilotInvalidRequestError( "Only image or non-source clips can be extended without media analysis." ) clip.duration_ms = duration_ms if isinstance(clip, SourceClip): clip.source_duration_ms = duration_ms async def _add_clip( self, document: EditorDocument, *, workspace_id: str, user_id: str, project_id: str, asset_id: str, duration_ms: int, ) -> None: asset = await self.assets.get_owned_by_id( workspace_id=workspace_id, user_id=user_id, asset_id=asset_id ) if asset.project_id != project_id: raise CopilotInvalidRequestError("The selected asset is not attached to this project.") mime = asset.mime_type if mime.startswith("audio/"): track_type = "audio" kind = "audio" elif mime.startswith("video/"): track_type = "video" kind = "video" elif mime.startswith("image/"): track_type = "video" kind = "image" else: raise CopilotInvalidRequestError("This asset type cannot be added to the timeline.") track = next( (item for item in document.timeline.tracks if item.type == track_type), None, ) if track is None: track = Track( id=str(uuid4()), type=track_type, name="Copilot media", order=len(document.timeline.tracks), muted=False, locked=False, visible=True, clips=[], ) document.timeline.tracks.append(track) metadata = asset.metadata_json or {} source_duration = metadata.get("duration_ms") if source_duration is None and isinstance(metadata.get("duration"), (int, float)): source_duration = round(float(metadata["duration"]) * 1_000) clip_duration = duration_ms if kind == "image" else source_duration if not isinstance(clip_duration, int) or clip_duration <= 0: raise CopilotInvalidRequestError("The asset has no validated duration metadata.") common = { "id": str(uuid4()), "trackId": track.id, "label": asset.filename[:500], "startMs": document.duration_ms(), "durationMs": clip_duration, "visible": True, "opacity": 1, "metadata": {}, "assetId": asset.id, "sourceStartMs": 0, "sourceDurationMs": clip_duration, } if kind == "audio": track.clips.append(AudioClip(**common, kind="audio", volume=1, fadeInMs=0, fadeOutMs=0)) else: track.clips.append( MediaClip( **common, kind="media", mediaType=kind, transform=ClipTransform(x=0, y=0, scaleX=1, scaleY=1, rotation=0), volume=1, ) ) @staticmethod def _success( action: CopilotAction, summary: str, resource_type: str, resource_id: str, ) -> CopilotActionResult: return CopilotActionResult( action_id=action.id, action_type=action.type, status="completed", summary=summary, resource_type=resource_type, resource_id=resource_id, )