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