personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""CLI command for interactive help with sol commands."""
5
6from __future__ import annotations
7
8import argparse
9import json
10import subprocess
11import sys
12
13from think.utils import setup_cli
14
15
16def _read_stdin() -> str:
17 """Read a question from stdin. Shows a prompt if running in a terminal."""
18 if sys.stdin.isatty():
19 print(
20 "Enter your question (Ctrl+D to submit):",
21 file=sys.stderr,
22 )
23 try:
24 return sys.stdin.read().strip()
25 except KeyboardInterrupt:
26 return ""
27
28
29def main() -> None:
30 """Entry point for ``sol help``."""
31 parser = argparse.ArgumentParser(
32 prog="sol help",
33 description="Get help with sol commands",
34 )
35 parser.add_argument(
36 "question",
37 nargs="*",
38 help="Question about sol commands",
39 )
40
41 args = setup_cli(parser)
42
43 if not args.question:
44 question = _read_stdin()
45 if not question:
46 # Imported here to avoid circular import (sol.py imports think.help_cli).
47 from sol import print_help
48
49 print_help()
50 return
51 else:
52 question = " ".join(args.question).strip()
53
54 config = {"name": "unified", "prompt": question}
55 config_json = json.dumps(config)
56
57 print("Thinking...", end="", file=sys.stderr, flush=True)
58
59 try:
60 proc = subprocess.Popen(
61 ["sol", "agents"],
62 stdin=subprocess.PIPE,
63 stdout=subprocess.PIPE,
64 stderr=subprocess.PIPE,
65 text=True,
66 )
67 except Exception as exc:
68 print(f"\rError: failed to run help agent: {exc}", file=sys.stderr)
69 sys.exit(1)
70
71 assert proc.stdin is not None # for type checker
72 proc.stdin.write(config_json + "\n")
73 proc.stdin.close()
74
75 finish_result: str | None = None
76 errors: list[str] = []
77
78 assert proc.stdout is not None # for type checker
79 for line in proc.stdout:
80 line = line.strip()
81 if not line:
82 continue
83
84 try:
85 event = json.loads(line)
86 except json.JSONDecodeError:
87 continue
88
89 event_type = event.get("event")
90 if event_type == "error":
91 errors.append(str(event.get("error", "Unknown error")))
92 elif event_type == "finish":
93 result_value = event.get("result")
94 finish_result = "" if result_value is None else str(result_value)
95
96 try:
97 proc.wait(timeout=120)
98 except subprocess.TimeoutExpired:
99 proc.kill()
100 print("\rError: help request timed out after 120 seconds.", file=sys.stderr)
101 sys.exit(1)
102
103 # Clear the "Thinking..." indicator.
104 print("\r \r", end="", file=sys.stderr, flush=True)
105
106 for message in errors:
107 print(f"Error: {message}", file=sys.stderr)
108
109 if finish_result is not None and finish_result.strip():
110 print(finish_result)
111 return
112
113 if finish_result is not None:
114 print("Error: help agent returned an empty result.", file=sys.stderr)
115 sys.exit(1)
116
117 stderr_output = proc.stderr.read() if proc.stderr else ""
118 if proc.returncode != 0 and stderr_output.strip():
119 print(f"Error: {stderr_output.strip()}", file=sys.stderr)
120 else:
121 print("Error: no help response received.", file=sys.stderr)
122 sys.exit(1)