"""Activity log: what Anomalous changed, newest first, kept in the ComfyUI user folder. Two sources feed it. The middleware records this plugin's own write requests (and Anomalous_TTS's) after they succeed: files such as models, covers, recipes, materials, notes, presets, images and audio. The browser posts the canvas changes it made (`POST /anomalous/activity`). Only the newest MAX_ENTRIES are kept and long values are shortened, so the file stays small. Recording never fails a request: the log is a convenience, not part of the action. """ import asyncio import json import logging import os import threading import time import uuid from aiohttp import web import folder_paths from .path_utils import atomic_write_json MAX_ENTRIES = 500 MAX_LOG_BYTES = 2 * 1024 * 1024 # long canvas entries could otherwise outgrow the JSON writer's limit MAX_VALUE_CHARS = 400 MAX_CHANGE_CHARS = 2000 # a before/after value: room for a whole prompt, so its receipt can compare tags DIFF_CONTEXT_CHARS = 60 # kept before the first difference of a long before/after pair MAX_CHANGES = 60 # Write routes worth a line in the log → the action id the browser has words for. Scans are # recorded when they end, with what they found (api/scan_report.py), not here. ROUTE_ACTIONS = { '/anomalous/delete_model': 'model_delete', '/anomalous/update_metadata': 'model_edit', '/anomalous/set_custom_cover': 'model_cover', '/anomalous/upload_custom_cover': 'model_cover', '/anomalous/save_recipe': 'recipe_save', '/anomalous/update_recipe': 'recipe_edit', '/anomalous/delete_recipe': 'recipe_delete', '/anomalous/restore_recipe_version': 'recipe_restore', '/anomalous/set_recipe_gallery_cover': 'recipe_cover', '/anomalous/import_recipe_package_commit': 'recipe_import', '/anomalous/save_image_material': 'material_save', '/anomalous/save_prompt_plan': 'material_save', '/anomalous/update_material': 'material_edit', '/anomalous/delete_material': 'material_delete', '/anomalous/save_notebook': 'note_save', '/anomalous/delete_notebook': 'note_delete', '/anomalous/save_parameter': 'preset_save', '/anomalous/rename_parameter': 'preset_rename', '/anomalous/delete_parameter': 'preset_delete', '/anomalous/delete_gallery_image': 'image_delete', '/anomalous/delete_audio_gallery': 'audio_delete', '/anomalous/save_audio_preview': 'audio_save', '/anomalous/clear_cache': 'cache_clear', '/anomalous_tts/import/commit': 'voice_import', '/anomalous_tts/settings': 'voice_settings', } # Request fields that name what was changed, in order of preference. TARGET_FIELDS = ('name', 'character', 'filename', 'target_filename', 'recipe_filename', 'notebook_filename', 'new_name', 'path', 'file', 'target') # What a handler attaches to its request for the log: request[ACTIVITY_DETAIL] = {fields, files}. # web.RequestKey is newer than the aiohttp some ComfyUI installs have; a plain key works there. ACTIVITY_DETAIL = web.RequestKey("activity_detail", dict) if hasattr(web, "RequestKey") else "activity_detail" _lock = threading.RLock() _entries = None _listeners = [] # called with each new entry once it is saved (ai_tasks.note_activity) def log_path(): user_dir = folder_paths.get_user_directory() if hasattr(folder_paths, 'get_user_directory') else None if not user_dir: user_dir = os.path.join(folder_paths.base_path, 'user') directory = os.path.join(user_dir, 'anomalous') os.makedirs(directory, exist_ok=True) return os.path.join(directory, 'activity_log.json') def _load(): global _entries if _entries is None: try: with open(log_path(), encoding='utf-8') as source: data = json.load(source) _entries = [entry for entry in data if isinstance(entry, dict)] if isinstance(data, list) else [] except (OSError, ValueError): _entries = [] return _entries def _encoded_size(entries): return len(json.dumps(entries, ensure_ascii=False, separators=(',', ':')).encode('utf-8')) def _trim(entries): """Drop the oldest entries until the list fits MAX_ENTRIES and MAX_LOG_BYTES.""" del entries[MAX_ENTRIES:] while len(entries) > 1 and _encoded_size(entries) > MAX_LOG_BYTES: del entries[max(1, len(entries) * 9 // 10):] def _text(value): return value if isinstance(value, str) else json.dumps(value, ensure_ascii=False) def _short(value, limit=MAX_VALUE_CHARS): text = _text(value) return text if len(text) <= limit else text[:limit] + '…' def _short_values(item, keys): """`keys` of `item` shortened. A long before/after pair is cut around its first difference instead of at the start, so a change late in a long prompt still shows.""" out = {key: _short(item[key], MAX_CHANGE_CHARS if key in ('before', 'after') else MAX_VALUE_CHARS) for key in keys if key in item and item[key] is not None} if 'before' not in out or 'after' not in out: return out old, new = _text(item['before']), _text(item['after']) if len(old) <= MAX_CHANGE_CHARS and len(new) <= MAX_CHANGE_CHARS: return out same = 0 while same < min(len(old), len(new)) and old[same] == new[same]: same += 1 start = max(0, same - DIFF_CONTEXT_CHARS) def cut(text): return ('…' if start else '') + text[start:start + MAX_CHANGE_CHARS] + ('…' if start + MAX_CHANGE_CHARS < len(text) else '') out['before'], out['after'] = cut(old), cut(new) return out def add_entry(source, action, target='', detail=None): """Append one entry (newest first) and save. Returns the entry.""" entry = { 'id': uuid.uuid4().hex[:12], 'time': time.time(), 'source': source, 'action': action, 'target': _short(target or ''), } if detail: entry['detail'] = detail with _lock: entries = _load() entries.insert(0, entry) _trim(entries) atomic_write_json(log_path(), entries) for listener in _listeners: try: listener(entry) except Exception: # noqa: BLE001 - a listener must never fail the action recorded logging.getLogger(__name__).warning('Anomalous activity log: a listener failed', exc_info=True) return entry def add_listener(listener): if listener not in _listeners: _listeners.append(listener) def tag(entry_id, task_id): """Names the AI task an entry belongs to (made by the AI, or by the user while it worked).""" with _lock: entries = _load() found = next((entry for entry in entries if entry.get('id') == entry_id), None) if found and found.get('task') != task_id: found['task'] = task_id atomic_write_json(log_path(), entries) def list_entries(limit=100, source=None): with _lock: entries = _load() picked = [entry for entry in entries if not source or entry.get('source') == source] return picked[:limit] def clear(): global _entries with _lock: _entries = [] atomic_write_json(log_path(), []) def _target_of(body): if not isinstance(body, dict): return '' for field in TARGET_FIELDS: value = body.get(field) if isinstance(value, str) and value.strip(): return os.path.basename(value.strip().rstrip('/\\')) or value.strip() return '' def _failed(response): """True when a response reports failure, including the 200 + {"status": "error"} style.""" if response.status >= 300: return True if getattr(response, 'content_type', '') != 'application/json' or not getattr(response, 'body', None): return False try: data = json.loads(response.body) except (TypeError, ValueError): return False return isinstance(data, dict) and (data.get('success') is False or data.get('ok') is False or data.get('status') in ('error', 'failed')) async def _request_body(request): """The body the handler already read (aiohttp caches it); {} when it was not JSON or a form.""" try: if request.content_type == 'application/json': return await request.json() if request.content_type in ('multipart/form-data', 'application/x-www-form-urlencoded'): return dict(await request.post()) except (ValueError, UnicodeDecodeError, RuntimeError): pass return {} def _clean_detail(detail): """What a handler attached for the log (request[ACTIVITY_DETAIL]): the fields it changed ({field, before, after}) and the files it moved, values shortened.""" if not isinstance(detail, dict): return None cleaned = {} if isinstance(detail.get('fields'), list): cleaned['fields'] = [_short_values(item, ('field', 'before', 'after')) for item in detail['fields'][:MAX_CHANGES] if isinstance(item, dict)] if isinstance(detail.get('files'), list): cleaned['files'] = [_short(name) for name in detail['files'][:MAX_CHANGES]] return {key: value for key, value in cleaned.items() if value} or None @web.middleware async def activity_middleware(request, handler): response = await handler(request) action = ROUTE_ACTIONS.get(request.path) if request.method == 'POST' else None if action and not _failed(response): try: target = _target_of(await _request_body(request)) detail = _clean_detail(request.get(ACTIVITY_DETAIL)) await asyncio.get_running_loop().run_in_executor(None, add_entry, 'file', action, target, detail) except Exception: # noqa: BLE001 - the log must never fail the action it records logging.getLogger(__name__).warning('Anomalous activity log: could not record %s', request.path, exc_info=True) return response def _clean_changes(changes): cleaned = [] for change in changes[:MAX_CHANGES]: if not isinstance(change, dict): continue item = _short_values(change, ('kind', 'node', 'type', 'widget', 'before', 'after')) if item.get('kind') in ('changed', 'wired', 'added', 'removed', 'opened'): cleaned.append(item) return cleaned async def api_get_activity(request): try: limit = max(1, min(MAX_ENTRIES, int(request.query.get('limit', '100')))) except ValueError: limit = 100 source = request.query.get('source') or None entries = await asyncio.get_running_loop().run_in_executor(None, list_entries, limit, source) return web.json_response({'entries': entries}) async def api_post_activity(request): """Canvas changes from the browser: {changes: [...], total, workflow, via?}; via "mcp" marks changes an AI app made through the MCP endpoint.""" try: body = await request.json() except ValueError: return web.json_response({'error': 'Invalid JSON body'}, status=400) if not isinstance(body, dict) or not isinstance(body.get('changes'), list): return web.json_response({'error': 'changes must be a list'}, status=400) changes = _clean_changes(body['changes']) if not changes: return web.json_response({'error': 'No recognizable changes'}, status=400) total = body.get('total') if isinstance(body.get('total'), int) else len(changes) detail = {'changes': changes, 'total': max(total, len(changes))} if body.get('via') == 'mcp': detail['via'] = 'mcp' entry = await asyncio.get_running_loop().run_in_executor( None, add_entry, 'canvas', 'canvas', body.get('workflow') if isinstance(body.get('workflow'), str) else '', detail) return web.json_response({'entry': entry}) async def api_clear_activity(request): await asyncio.get_running_loop().run_in_executor(None, clear) return web.json_response({'ok': True}) def register_routes(app): if hasattr(app, 'middlewares') and activity_middleware not in app.middlewares: app.middlewares.append(activity_middleware) app.router.add_get('/anomalous/activity', api_get_activity) app.router.add_post('/anomalous/activity', api_post_activity) app.router.add_post('/anomalous/activity/clear', api_clear_activity)