personal memory agent
Add --force option to sol import and fix SystemExit message display
- Fix sol.py run_command() to print SystemExit string messages to stderr
(previously string messages were silently swallowed)
- Add --force flag to importer that deletes existing import directory
and starts fresh, with clear error message guiding users to use it
- Move shutil import to top of file per project conventions
- Add test for string SystemExit handling
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
+41
-7
+8
-1
sol.py
+8
-1
sol.py
···
250
250
return 0
251
251
except SystemExit as e:
252
252
# Preserve exit code from subcommand
253
-
return e.code if isinstance(e.code, int) else (1 if e.code else 0)
253
+
# SystemExit can have int code, string message, or None
254
+
if isinstance(e.code, int):
255
+
return e.code
256
+
elif isinstance(e.code, str):
257
+
print(e.code, file=sys.stderr)
258
+
return 1
259
+
else:
260
+
return 0 if not e.code else 1
254
261
255
262
256
263
def main() -> None:
+12
tests/test_sol.py
+12
tests/test_sol.py
···
92
92
exit_code = sol.run_command("test.module")
93
93
assert exit_code == 1
94
94
95
+
def test_run_command_with_string_exit(self, capsys):
96
+
"""Test running a command that raises SystemExit with a string message."""
97
+
mock_module = MagicMock()
98
+
mock_module.main = MagicMock(side_effect=SystemExit("Error: something failed"))
99
+
100
+
with patch("importlib.import_module", return_value=mock_module):
101
+
exit_code = sol.run_command("test.module")
102
+
assert exit_code == 1
103
+
104
+
captured = capsys.readouterr()
105
+
assert "Error: something failed" in captured.err
106
+
95
107
def test_run_command_import_error(self):
96
108
"""Test handling ImportError for nonexistent module."""
97
109
with patch(
+21
-6
think/importer.py
+21
-6
think/importer.py
···
8
8
import os
9
9
import queue
10
10
import re
11
+
import shutil
11
12
import subprocess
12
13
import threading
13
14
import time
···
531
532
facet: str | None,
532
533
setting: str | None,
533
534
detection_result: dict | None,
535
+
force: bool = False,
534
536
) -> str:
535
537
"""Copy file to imports/ and write metadata. Returns new file path."""
536
538
journal_root = Path(get_journal())
···
538
540
539
541
# Check for conflict
540
542
if import_dir.exists():
541
-
raise SystemExit(
542
-
f"Error: Import already exists for timestamp {timestamp}\n"
543
-
f"Use a different timestamp or delete: {import_dir}"
544
-
)
543
+
if force:
544
+
logger.info(f"Removing existing import directory: {import_dir}")
545
+
shutil.rmtree(import_dir)
546
+
else:
547
+
raise SystemExit(
548
+
f"Error: Import already exists for timestamp {timestamp}\n"
549
+
f"To re-import, use --force to delete existing data and start over"
550
+
)
545
551
546
552
# Copy file to imports/
547
553
filename = os.path.basename(media_path)
···
620
626
action="store_true",
621
627
help="Skip waiting for transcription and summary generation",
622
628
)
629
+
parser.add_argument(
630
+
"--force",
631
+
action="store_true",
632
+
help="Force re-import by deleting existing import directory",
633
+
)
623
634
args, extra = setup_cli(parser, parse_known=True)
624
635
if extra and not args.timestamp:
625
636
args.timestamp = extra[0]
···
657
668
658
669
# Copy to imports/ if file is not already there
659
670
if needs_setup:
660
-
print("Copying to journal...")
661
671
args.media = _setup_import(
662
-
args.media, args.timestamp, args.facet, args.setting, detection_result
672
+
args.media,
673
+
args.timestamp,
674
+
args.facet,
675
+
args.setting,
676
+
detection_result,
677
+
force=args.force,
663
678
)
664
679
print("Starting import...")
665
680