solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""CLI for managing maintenance tasks.
5
6Usage:
7 sol maint # Run pending tasks
8 sol maint --list # Show status of all tasks
9 sol maint <task> # Show task details and log output
10 sol maint --force <task> # Re-run a specific task
11"""
12
13from __future__ import annotations
14
15import argparse
16import json
17import sys
18from datetime import datetime
19from pathlib import Path
20
21from think.utils import get_journal, setup_cli
22
23from .maint import (
24 get_state_file,
25 get_task_by_name,
26 get_task_status,
27 list_tasks,
28 run_pending_tasks,
29 run_task,
30)
31
32
33def _format_duration(ms: int) -> str:
34 """Format duration in milliseconds to a compact human-readable string."""
35 if ms < 1000:
36 return f"{ms}ms"
37 if ms < 60000:
38 return f"{ms // 1000}s"
39 return f"{ms // 60000}m {(ms % 60000) // 1000}s"
40
41
42def print_task(t: dict) -> None:
43 """Print a task summary line and optional run metadata."""
44 desc = f" - {t['description']}" if t["description"] else ""
45 status_info = ""
46 if t["status"] == "in_progress":
47 status_info = " (in progress)"
48 elif t["exit_code"] is not None and t["exit_code"] != 0:
49 status_info = f" (exit {t['exit_code']})"
50
51 print(f" {t['qualified_name']}{desc}{status_info}")
52
53 if t.get("ran_ts") is None:
54 return
55
56 ts_str = datetime.fromtimestamp(t["ran_ts"] / 1000).strftime("%Y-%m-%d %H:%M")
57 parts = [f"ran {ts_str}"]
58 detail_parts = []
59 if t.get("duration_ms") is not None:
60 detail_parts.append(_format_duration(t["duration_ms"]))
61 line_count = t.get("line_count", 0)
62 if line_count > 0:
63 detail_parts.append(f"{line_count} lines")
64 if detail_parts:
65 parts.append(f"({', '.join(detail_parts)})")
66
67 print(f" {' '.join(parts)}")
68
69
70def show_task_details(journal: Path, task_name: str) -> None:
71 """Show details and log output for a maintenance task."""
72 task = get_task_by_name(task_name)
73 if not task:
74 print(f"Task not found: {task_name}", file=sys.stderr)
75 print("Use 'sol maint --list' to see available tasks.", file=sys.stderr)
76 sys.exit(1)
77
78 status, exit_code, ran_ts = get_task_status(journal, task.app, task.name)
79 state_file = get_state_file(journal, task.app, task.name)
80
81 duration_ms = None
82 log_lines: list[str] = []
83 errors: list[str] = []
84 if status != "pending" and state_file.exists():
85 with open(state_file, "r") as f:
86 for raw_line in f:
87 line = raw_line.strip()
88 if not line:
89 continue
90 try:
91 event = json.loads(line)
92 except json.JSONDecodeError:
93 continue
94
95 event_type = event.get("event")
96 if event_type == "line":
97 text = event.get("line")
98 if isinstance(text, str):
99 log_lines.append(text)
100 elif event_type == "exit":
101 if isinstance(event.get("duration_ms"), int):
102 duration_ms = event["duration_ms"]
103 if event.get("error"):
104 errors.append(str(event["error"]))
105
106 print(task.qualified_name)
107 if task.description:
108 print(task.description)
109
110 if status == "pending":
111 print("Status: pending")
112 elif status == "in_progress":
113 print("Status: in progress")
114 elif status == "success":
115 print("Status: success (exit 0)")
116 elif exit_code is None:
117 print("Status: failed")
118 else:
119 print(f"Status: failed (exit {exit_code})")
120
121 if ran_ts is not None:
122 ts_str = datetime.fromtimestamp(ran_ts / 1000).strftime("%Y-%m-%d %H:%M")
123 if duration_ms is not None:
124 print(f"Ran: {ts_str} ({duration_ms}ms)")
125 else:
126 print(f"Ran: {ts_str}")
127
128 if state_file.exists():
129 print(f"Log: {state_file}")
130
131 print()
132
133 if status == "pending":
134 print("Task has not been run yet.")
135 return
136
137 for line in log_lines:
138 print(line)
139
140 for error in errors:
141 print(f"Error: {error}")
142
143
144def main() -> None:
145 """CLI entry point for sol maint command."""
146 parser = argparse.ArgumentParser(
147 description="Run maintenance tasks for apps",
148 formatter_class=argparse.RawDescriptionHelpFormatter,
149 epilog="""
150Examples:
151 sol maint Run all pending maintenance tasks
152 sol maint --list Show status of all tasks
153 sol maint chat:fix_x Show task details and log output
154 sol maint -f fix_x Re-run a specific task
155""",
156 )
157 parser.add_argument(
158 "task",
159 nargs="?",
160 help="Task to show details for (or to re-run with --force)",
161 )
162 parser.add_argument(
163 "--list",
164 "-l",
165 action="store_true",
166 help="List all tasks with their status",
167 )
168 parser.add_argument(
169 "--force",
170 "-f",
171 action="store_true",
172 help="Re-run a specific task (requires task name)",
173 )
174
175 args = setup_cli(parser)
176 journal = Path(get_journal())
177
178 # List mode
179 if args.list:
180 tasks = list_tasks(journal)
181 if not tasks:
182 print("No maintenance tasks found.")
183 return
184
185 # Group by status
186 pending = [t for t in tasks if t["status"] == "pending"]
187 in_progress = [t for t in tasks if t["status"] == "in_progress"]
188 success = [t for t in tasks if t["status"] == "success"]
189 failed = [t for t in tasks if t["status"] == "failed"]
190
191 if pending:
192 print(f"Pending ({len(pending)}):")
193 for t in pending:
194 print_task(t)
195
196 if in_progress:
197 print(f"In Progress ({len(in_progress)}):")
198 for t in in_progress:
199 print_task(t)
200
201 if failed:
202 print(f"Failed ({len(failed)}):")
203 for t in failed:
204 print_task(t)
205
206 if success:
207 print(f"Completed ({len(success)}):")
208 for t in success:
209 print_task(t)
210
211 return
212
213 # Force re-run a specific task
214 if args.force:
215 if not args.task:
216 print("--force requires a task name.", file=sys.stderr)
217 print("Usage: sol maint --force <task>", file=sys.stderr)
218 sys.exit(1)
219 task = get_task_by_name(args.task)
220 if not task:
221 print(f"Task not found: {args.task}", file=sys.stderr)
222 print("Use 'sol maint --list' to see available tasks.", file=sys.stderr)
223 sys.exit(1)
224 success, exit_code = run_task(journal, task)
225 sys.exit(0 if success else exit_code)
226
227 # Show task details
228 if args.task:
229 show_task_details(journal, args.task)
230 return
231
232 # Bare invocation - show in-progress, then run pending
233 tasks = list_tasks(journal)
234 in_progress = [t for t in tasks if t["status"] == "in_progress"]
235 if in_progress:
236 print(f"In Progress ({len(in_progress)}):")
237 for t in in_progress:
238 print_task(t)
239 print()
240
241 # Run pending tasks
242 ran, succeeded = run_pending_tasks(journal)
243 if ran == 0:
244 print("No pending maintenance tasks.")
245 else:
246 print(f"Completed {succeeded}/{ran} task(s)")
247 sys.exit(0 if succeeded == ran else 1)
248
249
250if __name__ == "__main__":
251 main()