solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""App plugin system context processors and helpers."""
5
6from __future__ import annotations
7
8from dataclasses import dataclass
9
10from flask import Flask, g, request, url_for
11
12from apps import AppRegistry
13
14
15def _get_facets_data() -> list[dict]:
16 """Get active facets data for templates."""
17 from think.facets import get_facets
18
19 from .config import apply_facet_order, load_convey_config
20
21 all_facets = get_facets()
22 facets_list = []
23
24 for name, data in all_facets.items():
25 if data.get("muted", False):
26 continue
27
28 facets_list.append(
29 {
30 "name": name,
31 "title": data.get("title", name),
32 "color": data.get("color", ""),
33 "emoji": data.get("emoji", ""),
34 }
35 )
36
37 # Apply custom ordering from config
38 config = load_convey_config()
39 return apply_facet_order(facets_list, config)
40
41
42def _get_selected_facet() -> str | None:
43 """Get selected facet from cookie, syncing with config.
44
45 Cookie takes precedence - if it differs from config, update config.
46 If no cookie exists, use config value as default.
47 Validates against active (non-muted) facets; stale values are cleared.
48 """
49 from .config import get_selected_facet, set_selected_facet
50
51 cookie_facet = request.cookies.get("selectedFacet")
52 config_facet = get_selected_facet()
53
54 # Empty string cookie -> treat as no selection, expire it
55 if cookie_facet == "":
56 set_selected_facet(None)
57 g.clear_facet_cookie = True
58 return None
59
60 # Resolve: cookie takes precedence
61 facet = cookie_facet if cookie_facet is not None else config_facet
62
63 # Validate against active (non-muted) facets
64 if facet:
65 active_names = {f["name"] for f in _get_facets_data()}
66 if facet not in active_names:
67 set_selected_facet(None)
68 g.clear_facet_cookie = True
69 return None
70
71 # Sync: cookie takes precedence, update config if different
72 if cookie_facet is not None and cookie_facet != config_facet:
73 set_selected_facet(cookie_facet)
74
75 return facet
76
77
78@dataclass
79class AttentionItem:
80 """A system attention item for the chat bar and triage context."""
81
82 placeholder_text: str
83 context_lines: list[str]
84
85
86def _resolve_attention(awareness_current: dict) -> AttentionItem | None:
87 """Check attention sources P0-P4, return highest priority or None."""
88 # P0: Cortex errors
89 try:
90 import json
91 from datetime import datetime
92 from pathlib import Path
93
94 from think.utils import get_journal
95
96 journal = Path(get_journal())
97 today = datetime.now().strftime("%Y%m%d")
98 day_index = journal / "agents" / f"{today}.jsonl"
99 if day_index.exists():
100 errors: dict[str, float] = {}
101 successes: dict[str, float] = {}
102 for line in day_index.read_text().splitlines():
103 if not line.strip():
104 continue
105 try:
106 entry = json.loads(line)
107 name = entry.get("name", "")
108 ts = entry.get("ts", 0)
109 if entry.get("status") == "error":
110 if ts > errors.get(name, 0):
111 errors[name] = ts
112 elif entry.get("status") == "completed":
113 if ts > successes.get(name, 0):
114 successes[name] = ts
115 except (json.JSONDecodeError, TypeError):
116 continue
117 unresolved = [
118 name
119 for name, err_ts in errors.items()
120 if successes.get(name, 0) <= err_ts
121 ]
122 if unresolved:
123 count = len(unresolved)
124 names = ", ".join(sorted(unresolved)[:3])
125 suffix = f" (+{count - 3} more)" if count > 3 else ""
126 placeholder = (
127 f"{count} agent error{'s' if count != 1 else ''} today"
128 " — ask what happened"
129 )
130 context = [
131 f"System health: {count} unresolved agent error(s) today: "
132 f"{names}{suffix}. If user asks what needs attention, "
133 "summarize which agents failed."
134 ]
135 return AttentionItem(
136 placeholder_text=placeholder,
137 context_lines=context,
138 )
139 except Exception:
140 pass
141
142 # P1: Capture stale
143 capture = awareness_current.get("capture", {})
144 if capture.get("status") == "stale":
145 placeholder = "Capture may be offline — ask me to check"
146 context = [
147 "System health: capture appears offline (observer heartbeats stale). "
148 "If user asks what needs attention, mention capture status."
149 ]
150 return AttentionItem(placeholder_text=placeholder, context_lines=context)
151
152 # P2: Recent import completion
153 imports = awareness_current.get("imports", {})
154 last_completed = imports.get("last_completed")
155 last_summary = imports.get("last_result_summary")
156 if last_completed and last_summary:
157 try:
158 from datetime import datetime, timedelta
159
160 completed_dt = datetime.fromisoformat(last_completed)
161 if datetime.now() - completed_dt < timedelta(hours=1):
162 placeholder = f"Import complete: {last_summary} — ask me about it"
163 if len(placeholder) > 90:
164 placeholder = "New import complete — ask me what arrived"
165 context = [
166 f"System health: import recently completed — {last_summary}. "
167 "If user asks what needs attention, mention the new import."
168 ]
169 return AttentionItem(
170 placeholder_text=placeholder,
171 context_lines=context,
172 )
173 except Exception:
174 pass
175
176 # P3: Daily analysis highlights
177 journal_state = awareness_current.get("journal", {})
178 if journal_state.get("first_daily_ready"):
179 try:
180 from datetime import datetime
181 from pathlib import Path
182
183 from think.utils import get_journal
184
185 journal = Path(get_journal())
186 today = datetime.now().strftime("%Y%m%d")
187 agents_dir = journal / today / "agents"
188 if agents_dir.is_dir():
189 outputs = sorted(p.stem for p in agents_dir.glob("*.md"))
190 if outputs:
191 count = len(outputs)
192 placeholder = (
193 f"{count} analysis report{'s' if count != 1 else ''} ready"
194 " — ask about your day"
195 )
196 context = [
197 f"System health: {count} daily analysis report(s) "
198 f"available today: {', '.join(outputs)}. User can ask "
199 "about any of these topics."
200 ]
201 return AttentionItem(
202 placeholder_text=placeholder,
203 context_lines=context,
204 )
205 except Exception:
206 pass
207
208 # P4: Owner voiceprint candidate ready for confirmation
209 voiceprint = awareness_current.get("voiceprint", {})
210 if voiceprint.get("status") == "candidate":
211 cluster_size = voiceprint.get("cluster_size", 0)
212 placeholder = "Voice pattern detected — confirm in Speakers"
213 context = [
214 f"System detected owner voice pattern from {cluster_size} voice samples. "
215 "Direct user to the Speakers app (/app/speakers) to confirm their voiceprint."
216 ]
217 return AttentionItem(placeholder_text=placeholder, context_lines=context)
218
219 return None
220
221
222def _resolve_placeholder(awareness_current: dict, day_count: int) -> str:
223 """Resolve chat bar placeholder text based on journal state."""
224 attention = _resolve_attention(awareness_current)
225 if attention:
226 return attention.placeholder_text
227 imports = awareness_current.get("imports", {})
228 if not imports.get("has_imported") and day_count < 3:
229 return (
230 "Bring in past conversations, calendar, or notes to give me context..."
231 )
232 if awareness_current.get("journal", {}).get("first_daily_ready"):
233 if day_count < 2:
234 return "Your first daily analysis is ready — ask me what I found..."
235 if day_count >= 7:
236 return (
237 "Ask me about your day, search your journal, or explore insights..."
238 )
239 return "Your daily analysis is ready — ask about today or anything in your journal..."
240 return "Capture is running — your first daily analysis will be ready soon..."
241
242
243def register_app_context(app: Flask, registry: AppRegistry) -> None:
244 """Register app system context processors and template filters."""
245 from .utils import DATE_RE, format_date_short
246
247 # Register Jinja2 filters
248 app.jinja_env.filters["format_date_short"] = format_date_short
249
250 @app.context_processor
251 def inject_app_context() -> dict:
252 """Inject app registry and facets context for new app system."""
253 from .config import apply_app_order, load_convey_config
254
255 # Parse URL path: /app/{app_name}/{day}/...
256 path_parts = request.path.split("/")
257
258 # Auto-extract app name from URL for /app/{app_name}/... routes
259 current_app_name = None
260 if (
261 len(path_parts) > 2
262 and path_parts[1] == "app"
263 and path_parts[2] in registry.apps
264 ):
265 current_app_name = path_parts[2]
266
267 # Auto-extract day from URL for apps with date_nav enabled
268 # Pattern: /app/{app_name}/{YYYYMMDD} or /app/{app_name}/{YYYYMMDD}/*
269 day = None
270 if (
271 current_app_name
272 and registry.apps[current_app_name].date_nav_enabled()
273 and len(path_parts) > 3
274 and DATE_RE.fullmatch(path_parts[3])
275 ):
276 day = path_parts[3]
277
278 facets = _get_facets_data()
279 selected_facet = _get_selected_facet()
280
281 # Build apps dict for menu-bar
282 apps_dict = {}
283 for app_instance in registry.apps.values():
284 apps_dict[app_instance.name] = {
285 "icon": app_instance.icon,
286 "label": app_instance.label,
287 }
288
289 # Apply custom ordering from config
290 config = load_convey_config()
291 apps_dict = apply_app_order(apps_dict, config)
292
293 # Override sol label if agent has a chosen name
294 if "sol" in apps_dict:
295 try:
296 from think.utils import get_config as _get_journal_config
297
298 journal_config = _get_journal_config()
299 agent_block = journal_config.get("agent", {})
300 if agent_block.get("name_status") in ("chosen", "self-named"):
301 agent_name = agent_block.get("name", "").strip()
302 if agent_name:
303 apps_dict["sol"]["label"] = agent_name
304 except Exception:
305 pass # Keep default label on any error
306
307 # Get starred apps list
308 starred_apps = config.get("apps", {}).get("starred", [])
309
310 # Chat bar placeholder based on journal state
311 chat_bar_placeholder = "Send a message..."
312 try:
313 from think.awareness import get_current
314 from think.utils import day_dirs
315
316 awareness_current = get_current()
317 day_count = len(day_dirs())
318 chat_bar_placeholder = _resolve_placeholder(awareness_current, day_count)
319 except Exception:
320 pass # Default placeholder on any error
321
322 return {
323 "app_registry": registry,
324 "app": current_app_name,
325 "apps": apps_dict,
326 "facets": facets,
327 "selected_facet": selected_facet,
328 "starred_apps": starred_apps,
329 "day": day,
330 "chat_bar_placeholder": chat_bar_placeholder,
331 }
332
333 @app.context_processor
334 def inject_vendor_helper() -> dict:
335 """Provide convenient vendor library helper for templates."""
336
337 def vendor_lib(library_name: str, file: str | None = None) -> str:
338 """Generate URL for vendor library.
339
340 Args:
341 library_name: Name of vendor library (e.g., 'marked')
342 file: Optional specific file, defaults to {library}.min.js
343
344 Returns:
345 URL to the vendor library file
346
347 Example:
348 {{ vendor_lib('marked') }}
349 → /static/vendor/marked/marked.min.js
350 """
351 if file is None:
352 file = f"{library_name}.min.js"
353 return url_for("static", filename=f"vendor/{library_name}/{file}")
354
355 return {"vendor_lib": vendor_lib}
356
357 @app.after_request
358 def clear_stale_facet_cookie(response):
359 if getattr(g, "clear_facet_cookie", False):
360 response.delete_cookie("selectedFacet", path="/", samesite="Lax")
361 return response