personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""Convey configuration management and API routes."""
5
6from __future__ import annotations
7
8import logging
9from pathlib import Path
10from typing import Any
11
12from flask import Blueprint, request
13
14from . import state
15from .utils import error_response, load_json, save_json, success_response
16
17logger = logging.getLogger(__name__)
18
19bp = Blueprint("config", __name__, url_prefix="/api/config")
20
21
22def _get_config_path() -> Path:
23 """Get path to config/convey.json in journal root."""
24 return Path(state.journal_root) / "config" / "convey.json"
25
26
27def load_convey_config() -> dict[str, Any]:
28 """Load config/convey.json from journal root.
29
30 Returns:
31 Config dict with optional fields:
32 - facets.order: list of facet names
33 - facets.selected: currently selected facet name or null
34 - apps.order: list of app names
35 Empty dict if file doesn't exist or can't be parsed.
36 """
37 config_path = _get_config_path()
38 data = load_json(config_path)
39 return data if isinstance(data, dict) else {}
40
41
42def save_convey_config(config: dict[str, Any]) -> bool:
43 """Save config/convey.json atomically.
44
45 Args:
46 config: Configuration dict to save
47
48 Returns:
49 True if successful, False otherwise
50 """
51 config_path = _get_config_path()
52
53 # Ensure config directory exists
54 config_path.parent.mkdir(parents=True, exist_ok=True)
55
56 return save_json(config_path, config, indent=2)
57
58
59def get_selected_facet() -> str | None:
60 """Get selected facet from config.
61
62 Returns:
63 Selected facet name, or None if not set
64 """
65 config = load_convey_config()
66 facets_config = config.get("facets", {})
67 return facets_config.get("selected")
68
69
70def set_selected_facet(facet: str | None) -> None:
71 """Update selected facet in config.
72
73 Args:
74 facet: Facet name to select, or None to clear selection
75 """
76 config = load_convey_config()
77
78 # Ensure facets section exists
79 if "facets" not in config:
80 config["facets"] = {}
81
82 # Update selected field
83 config["facets"]["selected"] = facet
84
85 # Save config (async safe - doesn't block if write fails)
86 success = save_convey_config(config)
87 if not success:
88 logger.warning(f"Failed to save selected facet: {facet}")
89
90
91def apply_facet_order(facets: list[dict], config: dict) -> list[dict]:
92 """Apply custom ordering from config to facet list.
93
94 Args:
95 facets: List of facet dicts with 'name' field
96 config: Config dict with optional facets.order field
97
98 Returns:
99 Reordered facet list (ordered items first, then alphabetical remainder)
100 """
101 order = config.get("facets", {}).get("order", [])
102 if not order:
103 return facets
104
105 # Create lookup by name
106 facet_map = {f["name"]: f for f in facets}
107
108 # Ordered items first (if they exist)
109 ordered = [facet_map[name] for name in order if name in facet_map]
110
111 # Remaining items alphabetically
112 ordered_names = set(order)
113 remaining = sorted(
114 [f for f in facets if f["name"] not in ordered_names],
115 key=lambda f: f["name"],
116 )
117
118 return ordered + remaining
119
120
121def apply_app_order(apps: dict[str, Any], config: dict) -> dict[str, Any]:
122 """Apply custom ordering from config to app dict.
123
124 Groups apps by starred status, then applies ordering within each group.
125 Starred apps appear first, followed by unstarred apps.
126
127 Args:
128 apps: Dict mapping app name to app data
129 config: Config dict with optional apps.order and apps.starred fields
130
131 Returns:
132 Reordered dict (starred apps first in order, then unstarred apps in order)
133 """
134 order = config.get("apps", {}).get("order", [])
135 starred = set(config.get("apps", {}).get("starred", []))
136
137 # Separate apps into starred and unstarred
138 starred_apps = {}
139 unstarred_apps = {}
140
141 for name, data in apps.items():
142 if name in starred:
143 starred_apps[name] = data
144 else:
145 unstarred_apps[name] = data
146
147 # Helper to order a subset of apps
148 def order_apps(app_dict: dict[str, Any], app_order: list[str]) -> dict[str, Any]:
149 ordered = {}
150 # Ordered items first (if they exist)
151 for name in app_order:
152 if name in app_dict:
153 ordered[name] = app_dict[name]
154 # Remaining items alphabetically
155 ordered_names = set(app_order)
156 for name in sorted(app_dict.keys()):
157 if name not in ordered_names:
158 ordered[name] = app_dict[name]
159 return ordered
160
161 # Order each group
162 ordered_starred = order_apps(starred_apps, order) if starred_apps else {}
163 ordered_unstarred = order_apps(unstarred_apps, order) if unstarred_apps else {}
164
165 # Combine: starred first, then unstarred
166 result = {}
167 result.update(ordered_starred)
168 result.update(ordered_unstarred)
169
170 return result
171
172
173def validate_config(config: dict[str, Any]) -> tuple[bool, str | None]:
174 """Validate config structure and values.
175
176 Args:
177 config: Config dict to validate
178
179 Returns:
180 (is_valid, error_message) tuple
181 """
182 # Check top-level structure
183 if not isinstance(config, dict):
184 return False, "Config must be a JSON object"
185
186 # Validate facets section if present
187 if "facets" in config:
188 facets_config = config["facets"]
189 if not isinstance(facets_config, dict):
190 return False, "facets must be an object"
191
192 # Validate facets.order
193 if "order" in facets_config:
194 order = facets_config["order"]
195 if not isinstance(order, list):
196 return False, "facets.order must be an array"
197 if not all(isinstance(name, str) for name in order):
198 return False, "facets.order must contain only strings"
199
200 # Validate facets.selected
201 if "selected" in facets_config:
202 selected = facets_config["selected"]
203 if selected is not None and not isinstance(selected, str):
204 return False, "facets.selected must be a string or null"
205
206 # Validate apps section if present
207 if "apps" in config:
208 apps_config = config["apps"]
209 if not isinstance(apps_config, dict):
210 return False, "apps must be an object"
211
212 # Validate apps.order
213 if "order" in apps_config:
214 order = apps_config["order"]
215 if not isinstance(order, list):
216 return False, "apps.order must be an array"
217 if not all(isinstance(name, str) for name in order):
218 return False, "apps.order must contain only strings"
219
220 # Validate apps.starred
221 if "starred" in apps_config:
222 starred = apps_config["starred"]
223 if not isinstance(starred, list):
224 return False, "apps.starred must be an array"
225 if not all(isinstance(name, str) for name in starred):
226 return False, "apps.starred must contain only strings"
227
228 return True, None
229
230
231# API Routes
232
233
234@bp.route("/convey")
235def get_config() -> tuple[Any, int]:
236 """GET /api/config/convey - Return current convey configuration.
237
238 Returns:
239 JSON response with config data
240 """
241 try:
242 config = load_convey_config()
243 return success_response({"config": config})
244 except Exception as e:
245 logger.error(f"Failed to load config: {e}", exc_info=True)
246 return error_response("Failed to load configuration", 500)
247
248
249@bp.route("/convey", methods=["POST"])
250def update_config() -> tuple[Any, int]:
251 """POST /api/config/convey - Update convey configuration.
252
253 Request body: Full or partial config object
254 {
255 "facets": {
256 "order": ["work", "personal"],
257 "selected": "work"
258 },
259 "apps": {
260 "order": ["home", "activities"]
261 }
262 }
263
264 Returns:
265 JSON success/error response
266 """
267 try:
268 # Parse request
269 new_config = request.get_json()
270 if not new_config:
271 return error_response("Request body must be JSON", 400)
272
273 # Validate structure
274 valid, error_msg = validate_config(new_config)
275 if not valid:
276 return error_response(f"Invalid config: {error_msg}", 400)
277
278 # Merge with existing config (partial updates supported)
279 current_config = load_convey_config()
280
281 # Deep merge facets section
282 if "facets" in new_config:
283 if "facets" not in current_config:
284 current_config["facets"] = {}
285 current_config["facets"].update(new_config["facets"])
286
287 # Deep merge apps section
288 if "apps" in new_config:
289 if "apps" not in current_config:
290 current_config["apps"] = {}
291 current_config["apps"].update(new_config["apps"])
292
293 # Save updated config
294 success = save_convey_config(current_config)
295 if not success:
296 return error_response("Failed to save configuration", 500)
297
298 return success_response({"config": current_config})
299
300 except Exception as e:
301 logger.error(f"Failed to update config: {e}", exc_info=True)
302 return error_response("Failed to update configuration", 500)
303
304
305@bp.route("/facets/order", methods=["POST"])
306def update_facet_order() -> tuple[Any, int]:
307 """POST /api/config/facets/order - Update facet ordering.
308
309 Request body: {"order": ["work", "personal", "research"]}
310
311 Returns:
312 JSON success/error response
313 """
314 try:
315 data = request.get_json()
316 if not data or "order" not in data:
317 return error_response("Request must include 'order' array", 400)
318
319 order = data["order"]
320 if not isinstance(order, list):
321 return error_response("'order' must be an array", 400)
322
323 if not all(isinstance(name, str) for name in order):
324 return error_response("'order' must contain only strings", 400)
325
326 # Load config and update facets.order
327 config = load_convey_config()
328 if "facets" not in config:
329 config["facets"] = {}
330 config["facets"]["order"] = order
331
332 # Save
333 success = save_convey_config(config)
334 if not success:
335 return error_response("Failed to save facet order", 500)
336
337 return success_response({"order": order})
338
339 except Exception as e:
340 logger.error(f"Failed to update facet order: {e}", exc_info=True)
341 return error_response("Failed to update facet order", 500)
342
343
344@bp.route("/apps/order", methods=["POST"])
345def update_app_order() -> tuple[Any, int]:
346 """POST /api/config/apps/order - Update app ordering.
347
348 Request body: {"order": ["home", "activities", "todos"]}
349
350 Returns:
351 JSON success/error response
352 """
353 try:
354 data = request.get_json()
355 if not data or "order" not in data:
356 return error_response("Request must include 'order' array", 400)
357
358 order = data["order"]
359 if not isinstance(order, list):
360 return error_response("'order' must be an array", 400)
361
362 if not all(isinstance(name, str) for name in order):
363 return error_response("'order' must contain only strings", 400)
364
365 # Load config and update apps.order
366 config = load_convey_config()
367 if "apps" not in config:
368 config["apps"] = {}
369 config["apps"]["order"] = order
370
371 # Save
372 success = save_convey_config(config)
373 if not success:
374 return error_response("Failed to save app order", 500)
375
376 return success_response({"order": order})
377
378 except Exception as e:
379 logger.error(f"Failed to update app order: {e}", exc_info=True)
380 return error_response("Failed to update app order", 500)
381
382
383@bp.route("/apps/star", methods=["POST"])
384def toggle_app_star() -> tuple[Any, int]:
385 """POST /api/config/apps/star - Toggle starred status of an app.
386
387 Request body: {"app": "activities", "starred": true}
388
389 Returns:
390 JSON success/error response
391 """
392 try:
393 data = request.get_json()
394 if not data or "app" not in data or "starred" not in data:
395 return error_response(
396 "Request must include 'app' and 'starred' fields", 400
397 )
398
399 app_name = data["app"]
400 starred = data["starred"]
401
402 if not isinstance(app_name, str):
403 return error_response("'app' must be a string", 400)
404
405 if not isinstance(starred, bool):
406 return error_response("'starred' must be a boolean", 400)
407
408 # Load config and update apps.starred
409 config = load_convey_config()
410 if "apps" not in config:
411 config["apps"] = {}
412
413 starred_apps = set(config["apps"].get("starred", []))
414
415 if starred:
416 starred_apps.add(app_name)
417 else:
418 starred_apps.discard(app_name)
419
420 config["apps"]["starred"] = sorted(starred_apps)
421
422 # Save
423 success = save_convey_config(config)
424 if not success:
425 return error_response("Failed to save app starred status", 500)
426
427 return success_response({"app": app_name, "starred": starred})
428
429 except Exception as e:
430 logger.error(f"Failed to toggle app star: {e}", exc_info=True)
431 return error_response("Failed to toggle app starred status", 500)
432
433
434@bp.route("/facets/select", methods=["POST"])
435def select_facet() -> tuple[Any, int]:
436 """POST /api/config/facets/select - Update selected facet.
437
438 Request body: {"facet": "work"} or {"facet": null}
439
440 Returns:
441 JSON success/error response
442 """
443 try:
444 data = request.get_json()
445 if data is None:
446 return error_response("Request body must be JSON", 400)
447
448 if "facet" not in data:
449 return error_response("Request must include 'facet' field", 400)
450
451 facet = data["facet"]
452 if facet is not None and not isinstance(facet, str):
453 return error_response("'facet' must be a string or null", 400)
454
455 # Update config
456 set_selected_facet(facet)
457
458 return success_response({"facet": facet})
459
460 except Exception as e:
461 logger.error(f"Failed to update selected facet: {e}", exc_info=True)
462 return error_response("Failed to update selected facet", 500)