rt/wrolpi
1
0
Fork 0
mirror of https://github.com/lrnselfreliance/wrolpi.git synced 2026-10-26 02:48:38 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

ai: ai.yaml config, model catalog, and the GGUF downloader

Browse source 
ai.yaml is the source of truth (opt-in: enabled=False by default) with
active model, idle-unload minutes, context override, and tool-call
bound.  The model catalog follows the map pattern: a bundled constant
keeps the Manage tab working offline, the CDN manifest (ai/models.json
+ detached GPG signature) wins when reachable.  GGUF downloads clone
the map-extract flow — the per-file meta4 is GPG-verified once and the
same bytes handed to aria2c, so the verified hash is the enforced one;
unverified downloads are refused.  Models land in <media>/ai/models/
(gated by the existing media_mounted download rules).  Manage endpoints
(catalog with RAM-tier recommendation, settings) live under
/api/ai/manage/ which the future agent loop excludes from its tool
catalog.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Roland 2026-08-25 21:21:34 -06:00
parent 85604c80fe
commit c6e0f18d5e
No known key found for this signature in database
GPG key ID: 1BA620CE6C9A4680
10 changed files with 593 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

4
main.py
View file

@ -289,6 +289,10 @@ async def start_single_tasks(app: Sanic):
from wrolpi.downloader import get_download_cache_config
get_download_cache_config().import_config()
logger.debug('download cache config imported')
with log_and_suppress(Exception, message='Failed to import ai config'):
from modules.ai.config import get_ai_config
get_ai_config().import_config()
logger.debug('ai config imported')
logger.debug('start_single_tasks waiting for db...')
async with flags.db_up.wait_for():

1
modules/ai/__init__.py
View file

@ -1 +1,2 @@
from modules.ai.api import ai_bp # noqa
from modules.ai.downloader import ModelDownloader # noqa Instantiation registers the downloader.

69
modules/ai/api.py
View file

@ -21,12 +21,13 @@ from wrolpi.api_utils import json_response
from wrolpi.collections.lib import search_collections
from wrolpi.common import api_param_limiter, get_media_directory, logger, wrol_mode_enabled
from wrolpi.downloader import download_manager
from wrolpi.errors import InvalidFile, SearchEmpty, UnknownFile
from wrolpi.errors import InvalidFile, SearchEmpty, UnknownFile, ValidationError
from wrolpi.files.lib import search_files
from wrolpi.vars import DOCKERIZED
from wrolpi.vars import DOCKERIZED, IS_RPI4
from wrolpi.version import __version__
from wrolpi.schema import JSONErrorResponse
from . import help_docs, lib, schema
from . import catalog, help_docs, lib, schema
from .config import get_ai_config
from .controller_client import controller_get
from .errors import ControllerUnavailable
@ -513,6 +514,68 @@ async def ai_list_disks(_: Request):
return json_response(ret)
# ---------------------------------------------------------------------------
# Manage endpoints — for the AI page's Manage tab, NOT part of the model's tool catalog.
# The agent loop excludes everything under /api/ai/manage/ when building tool definitions.
# ---------------------------------------------------------------------------
@ai_bp.get('/manage/catalog')
@openapi.description('The AI model catalog, download/active state, and the recommended tier for this'
' device. For the Manage tab, not the model.')
async def manage_catalog(_: Request):
models, source = await catalog.get_models_catalog()
models_directory = catalog.get_models_directory()
downloaded = {i.name for i in models_directory.glob('*.gguf')} if models_directory.is_dir() else set()
disk_usage = sum(i.stat().st_size for i in models_directory.glob('*')) if models_directory.is_dir() else 0
config = get_ai_config()
models = [dict(i, downloaded=i['name'] in downloaded, active=i['name'] == config.active_model)
for i in models]
total_ram = catalog.get_total_ram_bytes()
return json_response(dict(
models=models,
catalog_source=source,
total_ram=total_ram,
recommended_tier=catalog.recommend_tier(total_ram),
slow_hardware=IS_RPI4,
models_directory=str(models_directory),
disk_usage=disk_usage,
enabled=config.enabled,
active_model=config.active_model,
idle_unload_minutes=config.idle_unload_minutes,
context_size=config.context_size,
))
@ai_bp.post('/manage/settings')
@openapi.definition(
description='Update the AI settings (ai.yaml). For the Manage tab, not the model.',
body=schema.AIManageSettingsRequest,
)
@validate(schema.AIManageSettingsRequest)
async def manage_settings(_: Request, body: schema.AIManageSettingsRequest):
config = get_ai_config()
if body.active_model is not None:
if body.active_model and not (catalog.get_models_directory() / body.active_model).is_file():
raise ValidationError(f'Model is not downloaded: {body.active_model}')
config.active_model = body.active_model
if body.enabled is not None:
config.enabled = body.enabled
if body.idle_unload_minutes is not None:
if body.idle_unload_minutes < 1:
raise ValidationError('idle_unload_minutes must be at least 1')
config.idle_unload_minutes = body.idle_unload_minutes
if body.context_size is not None:
config.context_size = body.context_size
return json_response(dict(
enabled=config.enabled,
active_model=config.active_model,
idle_unload_minutes=config.idle_unload_minutes,
context_size=config.context_size,
))
@ai_bp.get('/files/read')
@openapi.definition(
summary='Read a text file',

108
modules/ai/catalog.py Normal file
View file

@ -0,0 +1,108 @@
"""The AI model catalog.
The bundled AI_MODELS constant is the offline fallback (like the map module's MAP_REGIONS);
the CDN manifest at ai/models.json is fetched at browse/download time and GPG-verified with the
committed WROLPi public key. The release process refreshes AI_MODELS from the published manifest.
"""
import json
import tempfile
from pathlib import Path
import psutil
from wrolpi.common import aiohttp_get, get_media_directory, logger, verify_gpg_signature
logger = logger.getChild(__name__)
CDN = 'https://wrolpi.nyc3.cdn.digitaloceanspaces.com'
MODELS_MANIFEST_URL = f'{CDN}/ai/models.json'
# Sizes are bytes; sha256 values are filled from the published manifest at release time. The
# meta4 sidecar (GPG-verified at download time) is what actually enforces the hash.
AI_MODELS = [
dict(
name='Qwen3-1.7B-Q4_K_M.gguf',
tier='small',
url=f'{CDN}/ai/Qwen3-1.7B-Q4_K_M.gguf',
sha256='',
size=1_200_000_000,
min_ram_gb=4,
default_context=8_192,
license='Apache 2.0',
description='The default assistant for 4GB devices (Pi 4/5). Best tool-calling per byte.',
),
dict(
name='Qwen3-4B-Instruct-2507-Q4_K_M.gguf',
tier='medium',
url=f'{CDN}/ai/Qwen3-4B-Instruct-2507-Q4_K_M.gguf',
sha256='',
size=2_600_000_000,
min_ram_gb=8,
default_context=16_384,
license='Apache 2.0',
description='The default assistant for 8GB+ devices. A standout small tool-caller.',
),
]
def get_models_directory() -> Path:
"""Where GGUF model files are stored. Always inside the media directory."""
return get_media_directory() / 'ai/models'
async def fetch_models_manifest(url: str = None) -> dict:
"""Fetch the models manifest from the CDN and verify its GPG signature (map-manifest pattern)."""
url = url or MODELS_MANIFEST_URL
sig_url = f'{url}.sig'
async with aiohttp_get(url, timeout=30) as response:
if response.status != 200:
raise RuntimeError(f'Failed to fetch models manifest: HTTP {response.status}')
manifest_bytes = await response.content.read()
async with aiohttp_get(sig_url, timeout=30) as response:
if response.status != 200:
raise RuntimeError(f'Failed to fetch models manifest signature: HTTP {response.status}')
signature_bytes = await response.content.read()
with tempfile.TemporaryDirectory() as tmpdir:
manifest_path = Path(tmpdir) / 'models.json'
sig_path = Path(tmpdir) / 'models.json.sig'
manifest_path.write_bytes(manifest_bytes)
sig_path.write_bytes(signature_bytes)
if not await verify_gpg_signature(manifest_path, sig_path):
raise RuntimeError('Models manifest GPG signature verification failed')
logger.info('Models manifest GPG signature verified successfully')
return json.loads(manifest_bytes)
async def get_models_catalog() -> (list, str):
"""The list of downloadable models, and where it came from ('cdn' or 'bundled').
The CDN manifest wins when reachable; the bundled constant keeps the Manage tab working
offline."""
try:
manifest = await fetch_models_manifest()
models = manifest.get('models') or []
if models:
return models, 'cdn'
except Exception as e:
logger.debug(f'Could not fetch models manifest, using bundled catalog: {e}')
return AI_MODELS, 'bundled'
def get_total_ram_bytes() -> int:
return psutil.virtual_memory().total
def recommend_tier(total_ram_bytes: int = None) -> str:
"""Recommend a model tier for this device's RAM. Other tiers stay selectable with a warning."""
total_ram_bytes = total_ram_bytes if total_ram_bytes is not None else get_total_ram_bytes()
gb = total_ram_bytes / 1024 ** 3
if gb < 6:
return 'small'
if gb < 24:
return 'medium'
return 'large'

101
modules/ai/config.py Normal file
View file

@ -0,0 +1,101 @@
"""The ai.yaml config — the source of truth for the AI assistant's state.
AI is opt-in: nothing is downloaded and llama-server does not run until the user enables it on
the Manage tab. YAML-only (no database)."""
import contextlib
from dataclasses import dataclass
from typing import Optional
from wrolpi.common import ConfigFile
@dataclass
class AIConfigValidator:
version: int = None
enabled: bool = None
active_model: str = None
idle_unload_minutes: int = None
context_size: Optional[int] = None
max_tool_calls: int = None
class AIConfig(ConfigFile):
file_name = 'ai.yaml'
default_config = dict(
version=0,
enabled=False,
active_model='', # a GGUF file name in <media>/ai/models/
idle_unload_minutes=15,
context_size=None, # None uses the active model's default from the catalog.
max_tool_calls=6,
)
validator = AIConfigValidator
def import_config(self, file=None, send_events=False):
super().import_config(file, send_events)
# YAML-only config (no database); a successful read is a successful import.
self.successful_import = True
@property
def enabled(self) -> bool:
return self._config['enabled']
@enabled.setter
def enabled(self, value: bool):
self.update({'enabled': bool(value)})
@property
def active_model(self) -> str:
return self._config['active_model']
@active_model.setter
def active_model(self, value: str):
self.update({'active_model': value or ''})
@property
def idle_unload_minutes(self) -> int:
return self._config['idle_unload_minutes']
@idle_unload_minutes.setter
def idle_unload_minutes(self, value: int):
self.update({'idle_unload_minutes': int(value)})
@property
def context_size(self) -> Optional[int]:
return self._config['context_size']
@context_size.setter
def context_size(self, value: Optional[int]):
self.update({'context_size': int(value) if value else None})
@property
def max_tool_calls(self) -> int:
return self._config['max_tool_calls']
@max_tool_calls.setter
def max_tool_calls(self, value: int):
self.update({'max_tool_calls': int(value)})
AI_CONFIG: AIConfig = AIConfig()
# Test override (see get_ai_config).
TEST_AI_CONFIG: Optional[AIConfig] = None
def get_ai_config() -> AIConfig:
global TEST_AI_CONFIG
if isinstance(TEST_AI_CONFIG, ConfigFile):
return TEST_AI_CONFIG
return AI_CONFIG
@contextlib.contextmanager
def ai_config_context():
"""Used to create a test config."""
global TEST_AI_CONFIG
TEST_AI_CONFIG = AIConfig()
try:
yield TEST_AI_CONFIG
finally:
TEST_AI_CONFIG = None

14
modules/ai/conftest.py
View file

@ -1,3 +1,17 @@
# AI module test fixtures. The docs conftest is not star-imported by the root conftest, but the
# AI endpoints cover docs; import its fixtures so AI tests can use doc_factory.
import pytest
from modules.ai.config import ai_config_context
from modules.docs.conftest import * # noqa
from wrolpi.api_utils import api_app
@pytest.fixture
def test_ai_config(test_directory, async_client):
"""An isolated ai.yaml config for one test."""
(test_directory / 'config').mkdir(exist_ok=True)
config_path = test_directory / 'config/ai.yaml'
with ai_config_context() as config:
config.initialize(api_app.shared_ctx.ai_config)
yield config_path

120
modules/ai/downloader.py Normal file
View file

@ -0,0 +1,120 @@
"""Downloads GGUF model files from the WROLPi CDN into <media>/ai/models/.
Follows the map-extract pattern: the per-file .meta4 sidecar is fetched once and its hash
signature GPG-verified, then the same bytes are handed to aria2c so the hash that was verified
is the hash that is enforced. Downloads are resumable and respect the media_mounted rules via
the download manager."""
import pathlib
from dataclasses import dataclass
from typing import Optional
from sqlalchemy.orm import Session
from modules.ai.catalog import get_models_directory
from modules.map.downloader import verify_meta4_signature
from wrolpi.common import logger
from wrolpi.downloader import Download, DownloadContext, Downloader, DownloadResult
from wrolpi.events import Events
__all__ = ['ModelDownloader', 'PreparedModel', 'ExecutedModel', 'ai_model_downloader']
logger = logger.getChild(__name__)
@dataclass
class PreparedModel:
"""Plan produced by ModelDownloader.prepare_download."""
name: str
models_directory: pathlib.Path
output_path: pathlib.Path
tmp_path: pathlib.Path
already_done: bool = False
error: Optional[str] = None
@dataclass
class ExecutedModel:
"""Output of ModelDownloader.execute_download."""
name: str
output_path: pathlib.Path
skipped: bool = False
error: Optional[str] = None
class ModelDownloader(Downloader):
"""Downloads one GGUF model file via aria2c with GPG-verified hash checking."""
name = 'ai_model'
listable = False
def prepare_download(self, session: Session, download: Download) -> PreparedModel:
name = download.url.rstrip('/').rsplit('/', 1)[-1]
models_directory = get_models_directory()
output_path = models_directory / name
tmp_path = models_directory / f'{name}.tmp'
if not name.endswith('.gguf'):
return PreparedModel(name=name, models_directory=models_directory, output_path=output_path,
tmp_path=tmp_path, error=f'Not a GGUF model URL: {download.url}')
prepared = PreparedModel(name=name, models_directory=models_directory, output_path=output_path,
tmp_path=tmp_path)
if output_path.is_file():
logger.info(f'{name} already exists, skipping download')
prepared.already_done = True
return prepared
models_directory.mkdir(parents=True, exist_ok=True)
# Clean up any leftover temp file; aria2c resumes its own .aria2 checkpoints.
tmp_path.unlink(missing_ok=True)
return prepared
async def execute_download(self, prepared: PreparedModel, ctx: DownloadContext,
download: Download = None) -> ExecutedModel:
if prepared.error:
return ExecutedModel(name=prepared.name, output_path=prepared.output_path, error=prepared.error)
if prepared.already_done:
return ExecutedModel(name=prepared.name, output_path=prepared.output_path, skipped=True)
url = download.url
# Fetch the meta4 once, GPG-verify the hash signature before downloading.
meta4_contents = await self.get_meta4_contents(url)
if not meta4_contents:
return ExecutedModel(name=prepared.name, output_path=prepared.output_path,
error=f'No meta4 available for {prepared.name}; refusing an unverified download')
if not await verify_meta4_signature(meta4_contents):
return ExecutedModel(name=prepared.name, output_path=prepared.output_path,
error=f'meta4 hash signature verification failed for {prepared.name}')
logger.info(f'meta4 signature verified for {prepared.name}')
downloaded_path = None
try:
downloaded_path = await self.download_file(
download, url, prepared.models_directory,
check_for_meta4=False, concurrent=1, meta4_xml=meta4_contents,
ctx=ctx,
)
downloaded_path.rename(prepared.tmp_path)
except Exception as e:
logger.error(f'Model download failed for {prepared.name}: {e}')
prepared.tmp_path.unlink(missing_ok=True)
if downloaded_path:
downloaded_path.unlink(missing_ok=True)
return ExecutedModel(name=prepared.name, output_path=prepared.output_path,
error=f'Download failed: {str(e)[:500]}')
# Replace atomically.
prepared.tmp_path.rename(prepared.output_path)
return ExecutedModel(name=prepared.name, output_path=prepared.output_path)
def finalize_download(self, session: Session, download: Download, executed: ExecutedModel) -> DownloadResult:
if executed.error:
return DownloadResult(success=False, error=executed.error)
if not executed.skipped:
Events.send_user_notify(f'AI model downloaded: {executed.name}', url='/ai/manage')
return DownloadResult(success=True, location='/ai/manage')
# Instantiating registers with the download manager.
ai_model_downloader = ModelDownloader()

8
modules/ai/schema.py
View file

@ -55,6 +55,14 @@ class AIHelpSearchRequest:
limit: int = 5
@dataclasses.dataclass
class AIManageSettingsRequest:
enabled: Optional[bool] = None
active_model: Optional[str] = None
idle_unload_minutes: Optional[int] = None
context_size: Optional[int] = None
@dataclasses.dataclass
class AISearchResponse:
results: list = dataclasses.field(default_factory=list)

163
modules/ai/test/test_config_catalog.py Normal file
View file

@ -0,0 +1,163 @@
"""Tests for the ai.yaml config, the model catalog, and the GGUF downloader."""
import json
from http import HTTPStatus
from unittest import mock
import pytest
import yaml
from modules.ai import catalog
from modules.ai.config import get_ai_config
from modules.ai.downloader import ai_model_downloader
from wrolpi.downloader import Download
def test_ai_config_defaults_and_update(test_ai_config):
"""The config has safe defaults (AI off) and round-trips values."""
config = get_ai_config()
assert config.enabled is False
assert config.active_model == ''
assert config.idle_unload_minutes == 15
assert config.context_size is None
assert config.max_tool_calls == 6
config.enabled = True
config.active_model = 'Qwen3-1.7B-Q4_K_M.gguf'
config.context_size = 8_192
assert get_ai_config().enabled is True
assert get_ai_config().active_model == 'Qwen3-1.7B-Q4_K_M.gguf'
assert get_ai_config().context_size == 8_192
@pytest.mark.asyncio
async def test_ai_config_import(test_ai_config):
"""A hand-edited ai.yaml is the source of truth on import."""
test_ai_config.write_text(yaml.dump(dict(
version=1, enabled=True, active_model='custom.gguf', idle_unload_minutes=5,
context_size=None, max_tool_calls=4,
)))
config = get_ai_config()
config.import_config(test_ai_config)
assert config.enabled is True
assert config.active_model == 'custom.gguf'
assert config.idle_unload_minutes == 5
assert config.max_tool_calls == 4
assert config.successful_import is True
@pytest.mark.asyncio
async def test_models_catalog_fallback(async_client):
"""The bundled catalog is served when the CDN is unreachable."""
with mock.patch('modules.ai.catalog.fetch_models_manifest', side_effect=RuntimeError('offline')):
models, source = await catalog.get_models_catalog()
assert source == 'bundled'
assert models == catalog.AI_MODELS
manifest = dict(version=1, models=[dict(name='new.gguf', tier='small')])
with mock.patch('modules.ai.catalog.fetch_models_manifest', return_value=manifest):
models, source = await catalog.get_models_catalog()
assert source == 'cdn'
assert models[0]['name'] == 'new.gguf'
def test_recommend_tier():
gb = 1024 ** 3
assert catalog.recommend_tier(4 * gb) == 'small'
assert catalog.recommend_tier(8 * gb) == 'medium'
assert catalog.recommend_tier(32 * gb) == 'large'
@pytest.mark.asyncio
async def test_model_downloader(test_session, test_directory, test_download_manager):
"""The model downloader verifies the meta4 signature, stages, and lands in <media>/ai/models."""
url = 'https://example.com/ai/Test-Model-Q4_K_M.gguf'
download = Download(url=url, downloader='ai_model')
prepared = ai_model_downloader.prepare_download(test_session, download)
assert prepared.error is None
assert prepared.output_path == test_directory / 'ai/models/Test-Model-Q4_K_M.gguf'
assert prepared.models_directory.is_dir()
meta4 = b'<metalink>fake</metalink>'
async def fake_download_file(download_, url_, directory, **kwargs):
assert kwargs['meta4_xml'] == meta4
path = directory / 'Test-Model-Q4_K_M.gguf'
path.write_bytes(b'GGUF fake weights')
return path
with mock.patch.object(ai_model_downloader, 'get_meta4_contents', return_value=meta4), \
mock.patch('modules.ai.downloader.verify_meta4_signature', return_value=True), \
mock.patch.object(ai_model_downloader, 'download_file', side_effect=fake_download_file):
executed = await ai_model_downloader.execute_download(prepared, ctx=None, download=download)
assert executed.error is None
assert prepared.output_path.is_file()
result = ai_model_downloader.finalize_download(test_session, download, executed)
assert result.success is True
# A second download of the same model is skipped.
prepared = ai_model_downloader.prepare_download(test_session, download)
assert prepared.already_done is True
executed = await ai_model_downloader.execute_download(prepared, ctx=None, download=download)
assert executed.skipped is True
@pytest.mark.asyncio
async def test_model_downloader_refuses_unverified(test_session, test_directory, test_download_manager):
"""No meta4, or a bad signature, refuses the download."""
url = 'https://example.com/ai/Test-Model.gguf'
download = Download(url=url, downloader='ai_model')
prepared = ai_model_downloader.prepare_download(test_session, download)
with mock.patch.object(ai_model_downloader, 'get_meta4_contents', return_value=None):
executed = await ai_model_downloader.execute_download(prepared, ctx=None, download=download)
assert 'refusing an unverified download' in executed.error
with mock.patch.object(ai_model_downloader, 'get_meta4_contents', return_value=b'<m/>'), \
mock.patch('modules.ai.downloader.verify_meta4_signature', return_value=False):
executed = await ai_model_downloader.execute_download(prepared, ctx=None, download=download)
assert 'signature verification failed' in executed.error
assert not prepared.output_path.exists()
# Not a GGUF URL is refused in prepare.
bad = ai_model_downloader.prepare_download(test_session, Download(url='https://example.com/x.zip'))
assert 'Not a GGUF model URL' in bad.error
@pytest.mark.asyncio
async def test_ai_manage_endpoints(async_client, test_directory, test_ai_config):
"""The Manage tab's catalog and settings endpoints."""
with mock.patch('modules.ai.catalog.fetch_models_manifest', side_effect=RuntimeError('offline')):
request, response = await async_client.get('/api/ai/manage/catalog')
assert response.status_code == HTTPStatus.OK
assert response.json['catalog_source'] == 'bundled'
assert response.json['enabled'] is False
assert response.json['recommended_tier'] in ('small', 'medium', 'large')
assert all(i['downloaded'] is False for i in response.json['models'])
# Selecting a model that is not downloaded is refused.
content = dict(active_model='Qwen3-1.7B-Q4_K_M.gguf')
request, response = await async_client.post('/api/ai/manage/settings', content=json.dumps(content))
assert response.status_code == HTTPStatus.BAD_REQUEST
# Once the file exists it can be selected, and shows downloaded/active in the catalog.
models_dir = test_directory / 'ai/models'
models_dir.mkdir(parents=True)
(models_dir / 'Qwen3-1.7B-Q4_K_M.gguf').write_bytes(b'GGUF')
content = dict(active_model='Qwen3-1.7B-Q4_K_M.gguf', enabled=True, idle_unload_minutes=5)
request, response = await async_client.post('/api/ai/manage/settings', content=json.dumps(content))
assert response.status_code == HTTPStatus.OK
assert response.json['active_model'] == 'Qwen3-1.7B-Q4_K_M.gguf'
assert response.json['enabled'] is True
with mock.patch('modules.ai.catalog.fetch_models_manifest', side_effect=RuntimeError('offline')):
request, response = await async_client.get('/api/ai/manage/catalog')
model = next(i for i in response.json['models'] if i['name'] == 'Qwen3-1.7B-Q4_K_M.gguf')
assert model['downloaded'] is True and model['active'] is True
assert response.json['disk_usage'] > 0
# Bad idle minutes.
request, response = await async_client.post('/api/ai/manage/settings',
content=json.dumps(dict(idle_unload_minutes=0)))
assert response.status_code == HTTPStatus.BAD_REQUEST

8
wrolpi/contexts.py
View file

@ -35,6 +35,7 @@ def attach_shared_contexts(app: Sanic):
app.shared_ctx.download_cache_config = manager.dict()
app.shared_ctx.map_pins_config = manager.dict()
app.shared_ctx.flasher_config = manager.dict()
app.shared_ctx.ai_config = manager.dict()
# Shared dicts.
app.shared_ctx.uploaded_files = manager.dict()
app.shared_ctx.status = manager.dict()
@ -126,6 +127,7 @@ def reset_shared_contexts(app: Sanic):
app.shared_ctx.domains_config.clear()
app.shared_ctx.archive_downloader_config.clear()
app.shared_ctx.download_cache_config.clear()
app.shared_ctx.ai_config.clear()
# Shared dicts.
app.shared_ctx.uploaded_files.clear()
app.shared_ctx.status.clear()
@ -313,3 +315,9 @@ def initialize_configs_contexts(app: Sanic):
FLASHER_CONFIG.initialize(app.shared_ctx.flasher_config)
except Exception as e:
logger.error(f'Failed to initialize in-memory flasher config: {e}')
try:
from modules.ai.config import AI_CONFIG
AI_CONFIG.initialize(app.shared_ctx.ai_config)
except Exception as e:
logger.error(f'Failed to initialize in-memory ai config: {e}')