personal memory agent
at main 122 lines 3.4 kB view raw
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)