personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""Utility functions for Convey app storage in journal."""
5
6import re
7from datetime import datetime
8from pathlib import Path
9from typing import Any
10
11from convey import state
12from think.utils import get_journal
13
14# Compiled pattern for app name validation
15APP_NAME_PATTERN = re.compile(r"^[a-z][a-z0-9_]*$")
16
17
18def get_app_storage_path(
19 app_name: str,
20 *sub_dirs: str,
21 ensure_exists: bool = True,
22) -> Path:
23 """
24 Get path to app storage directory in journal.
25
26 Args:
27 app_name: App name (must match [a-z][a-z0-9_]*)
28 *sub_dirs: Optional subdirectory components
29 ensure_exists: Create directory if it doesn't exist (default: True)
30
31 Returns:
32 Absolute path to <journal>/apps/<app_name>/<sub_dirs>/.
33 Falls back to think.utils.get_journal() when state.journal_root is empty.
34
35 Raises:
36 ValueError: If app_name contains invalid characters
37 RuntimeError: If the resolved journal root is not absolute
38
39 Examples:
40 get_app_storage_path("search") # → Path("<journal>/apps/search")
41 get_app_storage_path("search", "cache") # → Path("<journal>/apps/search/cache")
42 """
43 # Validate app_name to prevent path traversal
44 if not APP_NAME_PATTERN.match(app_name):
45 raise ValueError(f"Invalid app name: {app_name}")
46
47 # Build path
48 root = state.journal_root or get_journal()
49 if not Path(root).is_absolute():
50 raise RuntimeError(
51 f"get_app_storage_path: resolved journal root is not absolute: {root}"
52 )
53 path = Path(root) / "apps" / app_name
54 for sub_dir in sub_dirs:
55 path = path / sub_dir
56
57 if ensure_exists:
58 path.mkdir(parents=True, exist_ok=True)
59
60 return path
61
62
63def load_app_config(
64 app_name: str,
65 default: dict[str, Any] | None = None,
66) -> dict[str, Any] | None:
67 """
68 Load app configuration from <journal>/apps/<app_name>/config.json.
69
70 Args:
71 app_name: App name
72 default: Default value if config doesn't exist (default: None)
73
74 Returns:
75 Loaded JSON dict or default value if file doesn't exist
76
77 Examples:
78 config = load_app_config("my_app") # Returns None if missing
79 config = load_app_config("my_app", {}) # Returns {} if missing
80 """
81 from convey.utils import load_json
82
83 storage_path = get_app_storage_path(app_name, ensure_exists=False)
84 config_path = storage_path / "config.json"
85 return load_json(config_path) or default
86
87
88def save_app_config(
89 app_name: str,
90 config: dict[str, Any],
91) -> bool:
92 """
93 Save app configuration to <journal>/apps/<app_name>/config.json.
94
95 Args:
96 app_name: App name
97 config: Configuration dict to save
98
99 Returns:
100 True if successful, False otherwise
101 """
102 from convey.utils import save_json
103
104 storage_path = get_app_storage_path(app_name, ensure_exists=True)
105 config_path = storage_path / "config.json"
106 return save_json(config_path, config)
107
108
109def log_app_action(
110 app: str,
111 facet: str | None,
112 action: str,
113 params: dict[str, Any],
114 day: str | None = None,
115) -> None:
116 """Log a user-initiated action from a Convey app.
117
118 Creates a JSONL log entry for tracking user actions made through the web UI.
119
120 When facet is provided, writes to facets/{facet}/logs/{day}.jsonl.
121 When facet is None, writes to config/actions/{day}.jsonl for journal-level
122 actions (settings changes, observer management, etc.).
123
124 Args:
125 app: App name where action originated (e.g., "entities", "todos")
126 facet: Facet where action occurred, or None for journal-level actions
127 action: Action type (e.g., "entity_add", "todo_complete")
128 params: Action-specific parameters to record
129 day: Day in YYYYMMDD format (defaults to today)
130
131 Examples:
132 # Facet-scoped action
133 log_app_action(
134 app="entities",
135 facet="work",
136 action="entity_add",
137 params={"type": "Person", "name": "Alice"},
138 )
139
140 # Journal-level action (no facet)
141 log_app_action(
142 app="observer",
143 facet=None,
144 action="observer_create",
145 params={"name": "laptop"},
146 )
147 """
148 from think.facets import _write_action_log
149
150 if day is None:
151 day = datetime.now().strftime("%Y%m%d")
152
153 _write_action_log(
154 facet=facet,
155 action=action,
156 params=params,
157 source="app",
158 actor=app,
159 day=day,
160 )