solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4import json
5import re
6import time
7from datetime import datetime
8from pathlib import Path
9from typing import Any, Optional
10
11DATE_RE = re.compile(r"\d{8}")
12
13
14def format_date(date_str: str) -> str:
15 """Convert YYYYMMDD to 'Wednesday April 2nd' format."""
16 try:
17 date_obj = datetime.strptime(date_str, "%Y%m%d")
18 day = date_obj.day
19 if 10 <= day % 100 <= 20:
20 suffix = "th"
21 else:
22 suffix = {1: "st", 2: "nd", 3: "rd"}.get(day % 10, "th")
23 return date_obj.strftime(f"%A %B {day}{suffix}")
24 except ValueError:
25 return date_str
26
27
28def format_date_short(date_str: str) -> str:
29 """Convert YYYYMMDD to smart relative/short format.
30
31 Returns:
32 - "Today", "Yesterday", "Tomorrow" for those days
33 - Day name (e.g., "Wednesday") for dates within the past 6 days
34 - "Sat Nov 29" for other dates in current/recent year
35 - "Sat Nov 29 '24" for dates >6 months ago in a different year
36 """
37 try:
38 date_obj = datetime.strptime(date_str, "%Y%m%d")
39 today = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
40 date_normalized = date_obj.replace(hour=0, minute=0, second=0, microsecond=0)
41 delta_days = (date_normalized - today).days
42
43 # Today, Yesterday, Tomorrow
44 if delta_days == 0:
45 return "Today"
46 elif delta_days == -1:
47 return "Yesterday"
48 elif delta_days == 1:
49 return "Tomorrow"
50 # Within past 6 days - use day name
51 elif -6 <= delta_days < 0:
52 return date_obj.strftime("%A")
53 # Default short format
54 else:
55 short = date_obj.strftime("%a %b %-d")
56 # Add year suffix if >6 months ago AND different year
57 months_ago = (today.year - date_obj.year) * 12 + (
58 today.month - date_obj.month
59 )
60 if months_ago > 6 and date_obj.year != today.year:
61 short += date_obj.strftime(" '%y")
62 return short
63 except ValueError:
64 return date_str
65
66
67def time_since(epoch: int) -> str:
68 """Return short human readable age for ``epoch`` seconds."""
69 seconds = int(time.time() - epoch)
70 if seconds < 60:
71 return f"{seconds} seconds ago"
72 minutes = seconds // 60
73 if minutes < 60:
74 return f"{minutes} minute{'s' if minutes != 1 else ''} ago"
75 hours = minutes // 60
76 if hours < 24:
77 return f"{hours} hour{'s' if hours != 1 else ''} ago"
78 days = hours // 24
79 if days < 7:
80 return f"{days} day{'s' if days != 1 else ''} ago"
81 weeks = days // 7
82 return f"{weeks} week{'s' if weeks != 1 else ''} ago"
83
84
85def spawn_agent(
86 prompt: str,
87 name: str,
88 provider: Optional[str] = None,
89 config: Optional[dict[str, Any]] = None,
90) -> str | None:
91 """Spawn a Cortex agent and return the agent_id.
92
93 Thin wrapper around cortex_request that ensures imports are handled
94 and returns the agent_id directly.
95
96 Args:
97 prompt: The task or question for the agent
98 name: Agent name - system (e.g., "default") or app-qualified (e.g., "entities:entity_assist")
99 provider: Optional provider override (openai, google, anthropic)
100 config: Additional configuration (max_tokens, facet, session_id, etc.)
101
102 Returns:
103 agent_id string (timestamp-based), or None if the request could not be sent.
104
105 Raises:
106 ValueError: If config is invalid
107 """
108 from think.cortex_client import cortex_request
109
110 return cortex_request(
111 prompt=prompt,
112 name=name,
113 provider=provider,
114 config=config,
115 )
116
117
118def parse_pagination_params(
119 default_limit: int = 20,
120 max_limit: int = 100,
121 min_limit: int = 1,
122) -> tuple[int, int]:
123 """Parse and validate pagination parameters from request.args.
124
125 Extracts limit and offset from Flask request.args, validates them,
126 and enforces bounds to prevent API abuse.
127
128 Args:
129 default_limit: Default value for limit if not provided or invalid
130 max_limit: Maximum allowed value for limit
131 min_limit: Minimum allowed value for limit
132
133 Returns:
134 (limit, offset) tuple with validated integers
135
136 Example:
137 limit, offset = parse_pagination_params(default_limit=20, max_limit=100)
138 """
139 from flask import request
140
141 # Parse limit with error handling
142 try:
143 limit = int(request.args.get("limit", default_limit))
144 except (ValueError, TypeError):
145 limit = default_limit
146
147 # Parse offset with error handling
148 try:
149 offset = int(request.args.get("offset", 0))
150 except (ValueError, TypeError):
151 offset = 0
152
153 # Enforce bounds
154 limit = max(min_limit, min(limit, max_limit))
155 offset = max(0, offset)
156
157 return limit, offset
158
159
160def load_json(path: str | Path) -> dict | list | None:
161 """Load JSON file with consistent error handling.
162
163 Args:
164 path: Path to JSON file (string or Path object)
165
166 Returns:
167 Parsed JSON data (dict or list), or None if file doesn't exist or can't be parsed
168
169 Example:
170 data = load_json("config.json")
171 if data:
172 print(data.get("key"))
173 """
174 try:
175 with open(path, "r", encoding="utf-8") as f:
176 return json.load(f)
177 except (FileNotFoundError, json.JSONDecodeError, OSError):
178 return None
179
180
181def save_json(
182 path: str | Path,
183 data: dict | list,
184 indent: int = 2,
185 add_newline: bool = True,
186) -> bool:
187 """Save JSON file with consistent formatting.
188
189 Args:
190 path: Path to JSON file (string or Path object)
191 data: Data to serialize (dict or list)
192 indent: Indentation level (default: 2)
193 add_newline: Whether to add trailing newline for readability (default: True)
194
195 Returns:
196 True if successful, False otherwise
197
198 Example:
199 success = save_json("config.json", {"key": "value"})
200 """
201 try:
202 with open(path, "w", encoding="utf-8") as f:
203 json.dump(data, f, indent=indent, ensure_ascii=False)
204 if add_newline:
205 f.write("\n")
206 return True
207 except (OSError, TypeError):
208 return False
209
210
211def error_response(message: str, code: int = 400) -> tuple[Any, int]:
212 """Create a standard JSON error response.
213
214 Provides consistent error response format across all API endpoints.
215
216 Args:
217 message: Error message to return to client
218 code: HTTP status code (default: 400 Bad Request)
219
220 Returns:
221 Tuple of (jsonify response, status_code) ready for Flask return
222
223 Example:
224 return error_response("Invalid input", 400)
225 return error_response("Not found", 404)
226 """
227 from flask import jsonify
228
229 return jsonify({"error": message}), code
230
231
232def success_response(
233 data: dict[str, Any] | None = None, code: int = 200
234) -> tuple[Any, int]:
235 """Create a standard JSON success response.
236
237 Provides consistent success response format across all API endpoints.
238
239 Args:
240 data: Optional dict of additional data to include in response
241 code: HTTP status code (default: 200 OK)
242
243 Returns:
244 Tuple of (jsonify response, status_code) ready for Flask return
245
246 Example:
247 return success_response() # Returns {"success": True}
248 return success_response({"agent_id": "123"}) # Returns {"success": True, "agent_id": "123"}
249 """
250 from flask import jsonify
251
252 response_data = {"success": True}
253 if data:
254 response_data.update(data)
255 return jsonify(response_data), code