Spaces:
Paused
Paused
| """ | |
| Persona Customization API | |
| Handles RSI (avatar) upload and voice manifest customization. | |
| """ | |
| import os | |
| import json | |
| from pathlib import Path | |
| try: | |
| from flask import Request | |
| except Exception: | |
| from typing import Any as Request # type: ignore | |
| from python.helpers.api import ApiHandler | |
| import logging | |
| from datetime import datetime, timezone | |
| logger = logging.getLogger(__name__) | |
| class PersonaCustomization(ApiHandler): | |
| """Handler for persona customization operations.""" | |
| def requires_auth(cls) -> bool: | |
| return False | |
| def requires_csrf(cls) -> bool: | |
| return False | |
| def get_methods(cls) -> list[str]: | |
| return ["POST"] | |
| async def process(self, input: dict, request: Request | None = None) -> dict: | |
| """ | |
| Handle persona customization. | |
| Accepts: | |
| - rsi_upload: File upload for avatar image | |
| - rsi_mode: "generate", "upload", or "skip" | |
| - voice_preset: "generated", "minimal", "expressive", or "custom" | |
| - custom_tokens: Custom style tokens (when voice_preset="custom") | |
| Returns: | |
| Dictionary with status and updated paths. | |
| """ | |
| persona_dir = os.environ.get("AGENT_PROMPTS_DIR", "") | |
| persona_root = Path(persona_dir).parent if persona_dir else None | |
| # Enforce Emergence Gate edits policy if present (identity-only protection under TALK_ONLY) | |
| # Build list of fields being edited from explicit 'fields' key and any other indicators | |
| fields_being_edited = [] | |
| if isinstance(input.get("fields"), dict): | |
| fields_being_edited.extend(list(input.get("fields", {}).keys())) | |
| # infer field edits from higher-level inputs | |
| if "voice_preset" in input: | |
| fields_being_edited.append("voice_preset") | |
| if "rsi_mode" in input or (request and hasattr(request, "files") and "rsi_upload" in getattr(request, "files", {})): | |
| fields_being_edited.append("avatar") | |
| # Write an incoming debug record to aid diagnostics in tests | |
| try: | |
| if persona_root: | |
| dbg_in = persona_root / "persona_customize_incoming.jsonl" | |
| with open(dbg_in, "a", encoding="utf-8") as f: | |
| f.write(json.dumps({"ts": datetime.now(timezone.utc).isoformat(), "input": input, "fields_being_edited": fields_being_edited}) + "\n") | |
| except Exception: | |
| logger.exception("Failed to write incoming debug record") | |
| # Helper to append debug lines under persona root | |
| def _append_persona_debug(name: str, data: dict): | |
| try: | |
| if persona_root: | |
| dbg = persona_root / name | |
| with open(dbg, "a", encoding="utf-8") as f: | |
| f.write(json.dumps({"ts": datetime.now(timezone.utc).isoformat(), **data}) + "\n") | |
| except Exception: | |
| logger.exception("Failed to write persona debug %s", name) | |
| try: | |
| from python.helpers.emergence_check import edits_allowed_fields, get_gate_for_persona | |
| allowed, message, rejected = edits_allowed_fields(persona_root, fields_being_edited or None) | |
| override_used = False | |
| # If edits are not allowed, check for admin override | |
| if not allowed: | |
| gate = get_gate_for_persona(persona_root) | |
| override_payload_json = input.get("override_payload") | |
| override_signature = input.get("override_signature") | |
| if gate and override_payload_json and override_signature: | |
| logger.debug("PersonaCustomization: attempting admin override verification") | |
| parsed = gate.verify_admin_override(override_payload_json, override_signature) | |
| logger.debug("PersonaCustomization: override parsed=%s", parsed) | |
| # Write debug trace to persona root for easier test inspection | |
| try: | |
| if persona_root: | |
| dbg_path = persona_root / "persona_customize_debug.jsonl" | |
| with open(dbg_path, "a", encoding="utf-8") as dbg: | |
| dbg.write(json.dumps({ | |
| "ts": datetime.now(timezone.utc).isoformat(), | |
| "fields_being_edited": fields_being_edited, | |
| "override_payload_present": bool(override_payload_json), | |
| "override_signature_present": bool(override_signature), | |
| "parsed": parsed, | |
| }) + "\n") | |
| except Exception: | |
| logger.exception("Failed to write persona_customize debug") | |
| if parsed and parsed.get("action") == "override_edit": | |
| allowed_fields = parsed.get("fields", []) | |
| logger.debug("PersonaCustomization: override allowed_fields=%s", allowed_fields) | |
| if "*" in allowed_fields or all(f in allowed_fields for f in (fields_being_edited or [])): | |
| try: | |
| gate.record_override_usage(persona_root, fields_being_edited, parsed, override_signature) | |
| except Exception: | |
| logger.exception("Failed to record override usage") | |
| allowed = True | |
| rejected = [] | |
| override_used = True | |
| if not allowed: | |
| # Record rejected edit attempts for auditing | |
| try: | |
| if gate: | |
| gate.record_edit_rejection(persona_root, rejected or [], message or "edits disabled in TALK_ONLY") | |
| except Exception: | |
| logger.exception("Failed to record edit rejection") | |
| return {"error": message, "status": "forbidden", "rejected_fields": rejected} | |
| except Exception: | |
| # If emergence_check is unavailable, we proceed as before | |
| pass | |
| # Keep override usage flag available for downstream logic | |
| _override_used = override_used if 'override_used' in locals() else False | |
| # If edits were allowed (including via override), apply top-level 'fields' edits if present | |
| applied_fields = [] | |
| if input.get("fields") and isinstance(input.get("fields"), dict): | |
| try: | |
| config_path = persona_root / "persona_config.yaml" | |
| # Try to read existing YAML/JSON | |
| try: | |
| import yaml | |
| if config_path.exists(): | |
| with open(config_path, 'r', encoding='utf-8') as cf: | |
| config = yaml.safe_load(cf) or {} | |
| else: | |
| config = {} | |
| except Exception: | |
| # Fallback to JSON if PyYAML not available | |
| if config_path.exists(): | |
| with open(config_path, 'r', encoding='utf-8') as cf: | |
| config = json.load(cf) | |
| else: | |
| config = {} | |
| # Apply field edits | |
| for k, v in input.get("fields", {}).items(): | |
| config[k] = v | |
| applied_fields.append(k) | |
| # Write back using YAML if available | |
| try: | |
| import yaml | |
| with open(config_path, 'w', encoding='utf-8') as cf: | |
| yaml.safe_dump(config, cf) | |
| except Exception: | |
| with open(config_path, 'w', encoding='utf-8') as cf: | |
| json.dump(config, cf, indent=2) | |
| # Log applied edits | |
| try: | |
| gate = None | |
| try: | |
| from python.helpers.emergence_check import get_gate_for_persona | |
| gate = get_gate_for_persona(persona_root) | |
| except Exception: | |
| gate = None | |
| if gate: | |
| gate._append_event_log({"type": "edit_applied", "fields": applied_fields, "persona_root": str(persona_root)}) | |
| except Exception: | |
| logger.exception("Failed to log applied edits to emergence gate") | |
| except Exception: | |
| logger.exception("Failed to apply persona field edits") | |
| # If edits were allowed/override_used, ensure subsequent manifest updates do not re-check and block | |
| edits_globally_allowed = True if (fields_being_edited or []) and (not message or message is None) else False | |
| if override_used: | |
| edits_globally_allowed = True | |
| if not persona_root or not persona_root.exists(): | |
| return {"error": "Persona directory not found"} | |
| result = {"status": "success", "rsi_path": None, "voice_updated": False} | |
| # Handle RSI (Avatar) customization | |
| rsi_mode = input.get("rsi_mode", "skip") | |
| if rsi_mode == "upload": | |
| if "rsi_upload" in request.files: | |
| file = request.files["rsi_upload"] | |
| avatar_dir = persona_root / "persona_data" | |
| avatar_dir.mkdir(parents=True, exist_ok=True) | |
| avatar_path = avatar_dir / "avatar.png" | |
| file.save(str(avatar_path)) | |
| result["rsi_path"] = str(avatar_path) | |
| result["rsi_mode"] = "upload" | |
| elif rsi_mode == "generate": | |
| result["rsi_mode"] = "generate" | |
| result["message"] = "RSI will be generated from soul anchor during launch" | |
| else: | |
| result["rsi_mode"] = "skip" | |
| result["message"] = "Using default avatar placeholder" | |
| # Handle Voice Manifest customization | |
| voice_preset = input.get("voice_preset", "generated") | |
| tts_manifest_path = persona_root / "tts_voice_manifest.json" | |
| # Validate whether the requested fields are allowed under current gate state | |
| # If no specific fields provided in the input, we assume whole-file update (fields=None) | |
| fields_being_edited = None | |
| if isinstance(input.get("fields"), dict): | |
| fields_being_edited = list(input.get("fields", {}).keys()) | |
| # If edits are disallowed, edits_allowed_fields will return rejected fields and message | |
| try: | |
| from python.helpers.emergence_check import edits_allowed_fields, get_gate_for_persona | |
| allowed, message, rejected = edits_allowed_fields(persona_root, fields_being_edited) | |
| if not allowed: | |
| gate = get_gate_for_persona(persona_root) | |
| override_payload_json = input.get("override_payload") | |
| override_signature = input.get("override_signature") | |
| if gate and override_payload_json and override_signature: | |
| parsed = gate.verify_admin_override(override_payload_json, override_signature) | |
| if parsed and parsed.get("action") == "override_edit": | |
| allowed_fields = parsed.get("fields", []) | |
| if "*" in allowed_fields or all(f in allowed_fields for f in (fields_being_edited or [])): | |
| try: | |
| gate.record_override_usage(persona_root, fields_being_edited, parsed, override_signature) | |
| except Exception: | |
| logger.exception("Failed to record override usage") | |
| allowed = True | |
| rejected = [] | |
| if not allowed: | |
| # Record rejected edit attempts for auditing | |
| try: | |
| if gate: | |
| gate.record_edit_rejection(persona_root, rejected or [], message or "edits disabled in TALK_ONLY") | |
| except Exception: | |
| logger.exception("Failed to record edit rejection") | |
| return {"error": message, "status": "forbidden", "rejected_fields": rejected} | |
| except Exception: | |
| # if emergence_check is unavailable, we proceed as before | |
| pass | |
| if tts_manifest_path.exists(): | |
| try: | |
| with open(tts_manifest_path, 'r') as f: | |
| manifest = json.load(f) | |
| # Update voice preset | |
| if voice_preset in ["minimal", "expressive", "custom"]: | |
| manifest["voice_preset"] = voice_preset | |
| # If custom tokens provided | |
| if voice_preset == "custom" and "custom_tokens" in input: | |
| manifest["style_tokens"] = [token.strip() for token in input["custom_tokens"].split(",")] | |
| # Update engine if it was "generated" | |
| manifest["engine"] = "neutts-air" | |
| # Save updated manifest | |
| with open(tts_manifest_path, 'w') as f: | |
| json.dump(manifest, f, indent=2) | |
| result["voice_updated"] = True | |
| result["voice_preset"] = voice_preset | |
| except Exception as e: | |
| result["error"] = f"Failed to update voice manifest: {str(e)}" | |
| return result | |