my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	#!/usr/bin/env python3
#2	"""
#3	Mnemosyne CLI — v2
#4	==================
#5	Command-line interface for the Mnemosyne memory system.
#6	All commands use the v2 BEAM architecture (Mnemosyne/BeamMemory).
#7	"""
#8
#9	import os
#10	import sys
#11	import json
#12	from pathlib import Path
#13	from typing import NoReturn
#14
#15	# Data directory — respects MNEMOSYNE_DATA_DIR env var
#16	DATA_DIR = os.environ.get("MNEMOSYNE_DATA_DIR") or str(
#17	Path.home() / ".hermes" / "mnemosyne" / "data"
#18	)
#19	os.makedirs(DATA_DIR, exist_ok=True)
#20
#21
#22	def _fail(message: str, exit_code: int = 2) -> NoReturn:
#23	"""Print a CLI error and exit without a Python traceback."""
#24	print(f"Error: {message}", file=sys.stderr)
#25	raise SystemExit(exit_code)
#26
#27
#28	def _usage(message: str, exit_code: int = 2) -> NoReturn:
#29	"""Print command usage for invalid invocations and exit."""
#30	print(message, file=sys.stderr)
#31	raise SystemExit(exit_code)
#32
#33
#34	def _parse_float(value: str, name: str) -> float:
#35	"""Parse a float argument or exit with a user-facing CLI error."""
#36	try:
#37	return float(value)
#38	except ValueError:
#39	_fail(f"{name} must be a number: {value}")
#40
#41
#42	def _parse_int(value: str, name: str) -> int:
#43	"""Parse an integer argument or exit with a user-facing CLI error."""
#44	try:
#45	return int(value)
#46	except ValueError:
#47	_fail(f"{name} must be an integer: {value}")
#48
#49
#50	def _get_memory():
#51	"""Get a Mnemosyne v2 instance."""
#52	from mnemosyne.core.memory import Mnemosyne
#53	return Mnemosyne(db_path=os.path.join(DATA_DIR, "mnemosyne.db"))
#54
#55
#56	def cmd_store(args):
#57	"""Store a new memory."""
#58	if not args:
#59	_usage("Usage: mnemosyne store <content> [source] [importance]")
#60	content = args[0]
#61	source = args[1] if len(args) > 1 else "cli"
#62	importance = _parse_float(args[2], "importance") if len(args) > 2 else 0.5
#63
#64	mem = _get_memory()
#65	memory_id = mem.remember(
#66	content,
#67	source=source,
#68	importance=importance,
#69	extract_entities=True,
#70	)
#71	print(f"Stored: {memory_id}")
#72
#73
#74	def cmd_recall(args):
#75	"""Search memories."""
#76	if not args:
#77	_usage("Usage: mnemosyne recall <query> [top_k]")
#78	query = args[0]
#79	top_k = _parse_int(args[1], "top_k") if len(args) > 1 else 5
#80
#81	mem = _get_memory()
#82	results = mem.recall(query, top_k=top_k)
#83	print(f"\nResults for: {query}\n")
#84	for r in results:
#85	content = r.get("content", "")
#86	score = r.get("score", 0)
#87	print(f" ID: {r.get('id', '?')}")
#88	print(f" Content: {content[:150]}{'...' if len(content) > 150 else ''}")
#89	print(f" Score: {score:.3f}")
#90	if r.get("entity_match"):
#91	print(f" [entity match]")
#92	print()
#93
#94
#95	def cmd_update(args):
#96	"""Update an existing memory."""
#97	if len(args) < 2:
#98	_usage("Usage: mnemosyne update <memory_id> <new_content> [importance]")
#99	memory_id = args[0]
#100	content = args[1]
#101	importance = _parse_float(args[2], "importance") if len(args) > 2 else None
#102
#103	mem = _get_memory()
#104	success = mem.update(memory_id, content=content, importance=importance)
#105	if success:
#106	print(f"Updated: {memory_id}")
#107	else:
#108	_fail(f"Memory not found: {memory_id}", exit_code=1)
#109
#110
#111	def cmd_delete(args):
#112	"""Delete a memory."""
#113	if not args:
#114	_usage("Usage: mnemosyne delete <memory_id>")
#115	memory_id = args[0]
#116
#117	mem = _get_memory()
#118	success = mem.forget(memory_id)
#119	if success:
#120	print(f"Deleted: {memory_id}")
#121	else:
#122	_fail(f"Memory not found: {memory_id}", exit_code=1)
#123
#124
#125	def cmd_stats(args):
#126	"""Show memory system statistics."""
#127	mem = _get_memory()
#128	stats = mem.get_stats()
#129	beam = stats.get("beam", {})
#130	wm = beam.get("working_memory", {})
#131	ep = beam.get("episodic_memory", {})
#132	triples = beam.get("triples", {})
#133	print("\nMnemosyne Stats\n")
#134	print(f" Total memories: {stats.get('total_memories', 0)}")
#135	print(f" Working memory: {wm.get('total', 0)}")
#136	print(f" Episodic memory: {ep.get('total', 0)}")
#137	print(f" Knowledge triples: {triples.get('total', 0)}")
#138	if stats.get("banks"):
#139	print(f"\n Banks: {', '.join(stats['banks'])}")
#140	print(f" DB path: {stats.get('database', 'N/A')}")
#141
#142
#143	def cmd_sleep(args):
#144	"""Run consolidation cycle."""
#145	mem = _get_memory()
#146	# Use sleep_all_sessions to consolidate across ALL sessions, not just "default"
#147	# The per-session sleep() uses the Mnemosyne instance's session_id which is
#148	# always "default" when created from CLI — causing the phantom session bug.
#149	result = mem.sleep_all_sessions()
#150	print(f"Consolidation complete: {result}")
#151
#152
#153	def cmd_diagnose(args):
#154	"""Run PII-safe diagnostics."""
#155	try:
#156	from mnemosyne.diagnose import run_diagnostics
#157	result = run_diagnostics()
#158	print("\nMnemosyne Diagnostics\n")
#159	print(f" Checks passed: {result.get('checks_passed', 0)}/{result.get('checks_total', 0)}")
#160	if result.get("key_findings"):
#161	print("\n Key findings:")
#162	for finding in result["key_findings"]:
#163	print(f" - {finding}")
#164	else:
#165	print("\n No issues detected")
#166	except Exception as e:
#167	print(f"Diagnostic failed: {e}")
#168
#169
#170	def cmd_export(args):
#171	"""Export memories to JSON."""
#172	output_path = args[0] if args else os.path.join(DATA_DIR, "mnemosyne_export.json")
#173	mem = _get_memory()
#174	result = mem.export_to_file(output_path)
#175	print(
#176	"Exported "
#177	f"{result.get('working_memory_count', 0)} working, "
#178	f"{result.get('episodic_memory_count', 0)} episodic, "
#179	f"{result.get('legacy_memories_count', 0)} legacy, "
#180	f"{result.get('triples_count', 0)} triples "
#181	f"to {output_path}"
#182	)
#183
#184
#185	def cmd_import(args):
#186	"""Import memories from JSON."""
#187	if not args:
#188	_usage("Usage: mnemosyne import <file.json>")
#189	mem = _get_memory()
#190	try:
#191	result = mem.import_from_file(args[0])
#192	except FileNotFoundError:
#193	_fail(f"Import file not found: {args[0]}")
#194	except json.JSONDecodeError as e:
#195	_fail(f"Invalid JSON in import file {args[0]}: {e}")
#196	except ValueError as e:
#197	_fail(str(e))
#198	beam_stats = result.get("beam", {})
#199	print(
#200	"Imported "
#201	f"{beam_stats.get('working_memory', {}).get('inserted', 0)} working, "
#202	f"{beam_stats.get('episodic_memory', {}).get('inserted', 0)} episodic, "
#203	f"{result.get('legacy', {}).get('inserted', 0)} legacy, "
#204	f"{result.get('triples', {}).get('inserted', 0)} triples "
#205	f"from {args[0]}"
#206	)
#207
#208
#209	def cmd_import_hindsight(args):
#210	"""Import memories from a Hindsight JSON export or API."""
#211	if not args:
#212	_usage("Usage: mnemosyne import-hindsight <file.json\|base_url> [bank]")
#213	target = args[0]
#214	bank = args[1] if len(args) > 1 else "hermes"
#215	mem = _get_memory()
#216	from mnemosyne.core.importers.hindsight import import_from_hindsight
#217	if target.startswith("http://") or target.startswith("https://"):
#218	result = import_from_hindsight(mem, base_url=target, bank=bank)
#219	else:
#220	result = import_from_hindsight(mem, file_path=target, bank=bank)
#221	print(result.to_json())
#222	if result.errors:
#223	raise SystemExit(1)
#224
#225
#226	def cmd_mcp(args):
#227	"""Start MCP server."""
#228	try:
#229	from mnemosyne.mcp_server import main as mcp_main
#230	mcp_main(args)
#231	except ImportError:
#232	print("MCP not available. Install with: pip install mnemosyne-memory[mcp]")
#233	sys.exit(1)
#234
#235
#236	def cmd_clawd(args):
#237	"""Run OpenClawd brain commands."""
#238	from mnemosyne.clawd_brain import run_cli as clawd_run_cli
#239	clawd_run_cli(args)
#240
#241
#242	def cmd_bank(args):
#243	"""Manage memory banks."""
#244	if not args:
#245	_usage("Usage: mnemosyne bank <list\|create\|delete> [name]")
#246
#247	from mnemosyne.core.banks import BankManager
#248	bm = BankManager(Path(DATA_DIR))
#249
#250	subcmd = args[0]
#251	try:
#252	if subcmd == "list":
#253	banks = bm.list_banks()
#254	print("\nMemory Banks:\n")
#255	for b in banks:
#256	print(f" - {b}")
#257	elif subcmd == "create":
#258	if len(args) < 2:
#259	_fail("Usage: mnemosyne bank create <name>")
#260	bm.create_bank(args[1])
#261	print(f"Created bank: {args[1]}")
#262	elif subcmd == "delete":
#263	if len(args) < 2:
#264	_fail("Usage: mnemosyne bank delete <name>")
#265	if bm.delete_bank(args[1]):
#266	print(f"Deleted bank: {args[1]}")
#267	else:
#268	_fail(f"Bank not found: {args[1]}", exit_code=1)
#269	else:
#270	_fail(f"Unknown bank command: {subcmd}")
#271	except ValueError as e:
#272	_fail(str(e))
#273
#274
#275	COMMANDS = {
#276	"store": cmd_store,
#277	"remember": cmd_store,
#278	"recall": cmd_recall,
#279	"search": cmd_recall,
#280	"update": cmd_update,
#281	"edit": cmd_update,
#282	"delete": cmd_delete,
#283	"forget": cmd_delete,
#284	"stats": cmd_stats,
#285	"sleep": cmd_sleep,
#286	"consolidate": cmd_sleep,
#287	"diagnose": cmd_diagnose,
#288	"export": cmd_export,
#289	"import": cmd_import,
#290	"import-hindsight": cmd_import_hindsight,
#291	"mcp": cmd_mcp,
#292	"clawd": cmd_clawd,
#293	"clawd-brain": cmd_clawd,
#294	"bank": cmd_bank,
#295	}
#296
#297
#298	def run_cli():
#299	"""Main CLI entry point."""
#300	if len(sys.argv) < 2 or sys.argv[1] in ("--help", "-h", "help"):
#301	print("Mnemosyne — Local AI Memory System\n")
#302	print("Usage: mnemosyne <command> [args]\n")
#303	print("Commands:")
#304	print(" store <content> [source] [importance] Store a memory")
#305	print(" recall <query> [top_k] Search memories")
#306	print(" update <id> <content> [importance] Update a memory")
#307	print(" delete <id> Delete a memory")
#308	print(" stats Show statistics")
#309	print(" sleep Run consolidation")
#310	print(" diagnose Run diagnostics")
#311	print(" export [file.json] Export memories")
#312	print(" import <file.json> Import memories")
#313	print(" import-hindsight <file\|url> [bank] Import Hindsight memories")
#314	print(" clawd <init\|remember\|recall\|research\|ingest-ooda\|status>")
#315	print(" OpenClawd persistent brain/wiki")
#316	print(" bank list\|create\|delete [name] Manage memory banks")
#317	print(" mcp [--transport sse] [--port 8080] Start MCP server")
#318	return
#319
#320	command = sys.argv[1]
#321	handler = COMMANDS.get(command)
#322
#323	if handler:
#324	handler(sys.argv[2:])
#325	else:
#326	print(f"Unknown command: {command}", file=sys.stderr)
#327	print("Run 'mnemosyne --help' for usage.", file=sys.stderr)
#328	raise SystemExit(2)
#329
#330
#331	if __name__ == "__main__":
#332	run_cli()
#333

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public