from __future__ import annotations import argparse import json import random import time from pathlib import Path from .engine import MemoryEngine from .http_api import run_http_server from .service import MemoryService def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="shared-agent-memory") parser.add_argument("--db", default="shared-agent-memory.sqlite3") subparsers = parser.add_subparsers(dest="command", required=True) subparsers.add_parser("init-db") project = subparsers.add_parser("create-project") project.add_argument("project_id") project.add_argument("name") project.add_argument("--description") repo = subparsers.add_parser("create-repo") repo.add_argument("repo_id") repo.add_argument("name") repo.add_argument("--project-id") repo.add_argument("--path") profile = subparsers.add_parser("upsert-profile-fact") profile.add_argument("title") profile.add_argument("content") profile.add_argument("--source-ref") profile.add_argument("--evidence-ref") ingest = subparsers.add_parser("ingest") ingest.add_argument("payload_json") capture = subparsers.add_parser("capture-conversation") capture.add_argument("source_ref") capture.add_argument("scope") capture.add_argument("content") capture.add_argument("--capture-mode", default="manual_import") session = subparsers.add_parser("capture-session") session.add_argument("source_ref") session.add_argument("scope") session.add_argument("content") process = subparsers.add_parser("process-pending") process.add_argument("--limit", type=int) maintenance_tick = subparsers.add_parser("maintenance-tick") maintenance_tick.add_argument("--limit", type=int) maintenance_tick.add_argument("--job-id") maintenance_status = subparsers.add_parser("maintenance-status") maintenance_status.add_argument("--runs", type=int, default=10) maintenance_config = subparsers.add_parser("maintenance-config") maintenance_config.add_argument("job_id") maintenance_config.add_argument("--cadence") maintenance_config.add_argument("--interval-minutes", type=int) maintenance_config.add_argument("--window-start") maintenance_config.add_argument("--window-end") maintenance_config.add_argument("--enable", action="store_true") maintenance_config.add_argument("--disable", action="store_true") maintenance_config.add_argument("--metadata-json") maintenance_config.add_argument("--recompute-next", action="store_true") maintenance_daemon = subparsers.add_parser("maintenance-daemon") maintenance_daemon.add_argument("--interval-seconds", type=int, default=900) maintenance_daemon.add_argument("--jitter-seconds", type=int, default=60) maintenance_daemon.add_argument("--limit", type=int) maintenance_daemon.add_argument("--job-id") maintenance_daemon.add_argument("--once", action="store_true") serve_http = subparsers.add_parser("serve-http") serve_http.add_argument("--host", default="127.0.0.1") serve_http.add_argument("--port", type=int, default=8091) search = subparsers.add_parser("search") search.add_argument("query") search.add_argument("--scopes", nargs="*") search.add_argument("--project-id") search.add_argument("--repo-id") search.add_argument("--type") search.add_argument("--include-inbox", action="store_true") context = subparsers.add_parser("context") context.add_argument("--project") context.add_argument("--repo") context.add_argument("--agent") context.add_argument("--task") consolidate = subparsers.add_parser("consolidate") consolidate.add_argument("memory_ids", nargs="+") explain = subparsers.add_parser("explain") explain.add_argument("identifier") list_memories = subparsers.add_parser("list-memories") list_memories.add_argument("--status", default="active") list_memories.add_argument("--scope") list_memories.add_argument("--type") list_memories.add_argument("--subtype") list_memories.add_argument("--project-id") list_memories.add_argument("--repo-id") list_memories.add_argument("--run-id") list_memories.add_argument("--task-id") list_memories.add_argument("--origin-agent") list_memories.add_argument("--url") list_memories.add_argument("--domain") list_memories.add_argument("--source-ref") list_memories.add_argument("--evidence-ref") list_memories.add_argument("--metadata", action="append", default=[]) list_memories.add_argument("--limit", type=int, default=50) task_bundle = subparsers.add_parser("task-bundle") task_bundle.add_argument("task_id") subparsers.add_parser("audit-v2") subparsers.add_parser("migrate-v2") export = subparsers.add_parser("export") export.add_argument("--output") return parser def main() -> None: parser = build_parser() args = parser.parse_args() service = MemoryService(args.db) engine = MemoryEngine(service) if args.command == "init-db": print(json.dumps({"db": str(Path(args.db).resolve()), "status": "initialized"}, indent=2)) elif args.command == "create-project": service.create_project(args.project_id, args.name, args.description) print(json.dumps({"project_id": args.project_id}, indent=2)) elif args.command == "create-repo": service.create_repo(args.repo_id, args.name, args.project_id, args.path) print(json.dumps({"repo_id": args.repo_id}, indent=2)) elif args.command == "upsert-profile-fact": print( json.dumps( service.upsert_profile_fact(args.title, args.content, args.evidence_ref, args.source_ref), indent=2, ) ) elif args.command == "ingest": print(json.dumps(service.ingest(json.loads(args.payload_json)), indent=2)) elif args.command == "capture-conversation": print( json.dumps( service.capture_conversation(args.source_ref, args.scope, args.content, capture_mode=args.capture_mode), indent=2, ) ) elif args.command == "capture-session": print(json.dumps(service.capture_session(args.source_ref, args.scope, args.content), indent=2)) elif args.command == "process-pending": print(json.dumps(engine.process_pending(args.limit), indent=2)) elif args.command == "maintenance-tick": print(json.dumps(engine.run_due_maintenance(limit=args.limit, job_id=args.job_id), indent=2)) elif args.command == "maintenance-status": print( json.dumps( {"jobs": service.list_maintenance_jobs(), "runs": service.list_maintenance_runs(limit=args.runs)}, indent=2, ) ) elif args.command == "maintenance-config": job = service.get_maintenance_job(args.job_id) metadata = None if args.metadata_json: metadata = json.loads(args.metadata_json) enabled = None if args.enable: enabled = True if args.disable: enabled = False cadence = args.cadence or job["cadence"] interval_minutes = args.interval_minutes if args.interval_minutes is not None else job.get("interval_minutes") window_start = args.window_start if args.window_start is not None else job.get("window_start") window_end = args.window_end if args.window_end is not None else job.get("window_end") next_due_at = None if args.recompute_next: next_due_at = service.compute_next_due( cadence=cadence, interval_minutes=interval_minutes, window_start=window_start, window_end=window_end, ) updated = service.update_maintenance_job( args.job_id, cadence=args.cadence, interval_minutes=args.interval_minutes, window_start=args.window_start, window_end=args.window_end, next_due_at=next_due_at, enabled=enabled, metadata=metadata, ) print(json.dumps(updated, indent=2)) elif args.command == "maintenance-daemon": def _tick() -> dict[str, object]: results = engine.run_due_maintenance(limit=args.limit, job_id=args.job_id) return {"timestamp": time.time(), "results": results} print(json.dumps(_tick(), indent=2)) if args.once: return while True: jitter = random.randint(0, max(0, args.jitter_seconds)) if args.jitter_seconds else 0 time.sleep(max(1, args.interval_seconds) + jitter) print(json.dumps(_tick(), indent=2)) elif args.command == "serve-http": run_http_server(db_path=args.db, host=args.host, port=args.port) elif args.command == "search": filters = { key: value for key, value in { "project_id": args.project_id, "repo_id": args.repo_id, "type": args.type, }.items() if value } print(json.dumps(service.search(args.query, args.scopes, filters, include_inbox=args.include_inbox), indent=2)) elif args.command == "context": print(json.dumps(service.context_for(args.project, args.repo, args.agent, args.task), indent=2)) elif args.command == "consolidate": print(json.dumps(service.consolidate(args.memory_ids), indent=2)) elif args.command == "explain": print(json.dumps(service.explain(args.identifier), indent=2)) elif args.command == "list-memories": metadata = {} for pair in args.metadata: key, _, value = pair.partition("=") if not key or not _: raise ValueError(f"Invalid metadata filter: {pair}") metadata[key] = value print( json.dumps( service.list_memories( status=args.status, scope=args.scope, memory_type=args.type, subtype=args.subtype, project_id=args.project_id, repo_id=args.repo_id, run_id=args.run_id, task_id=args.task_id, origin_agent=args.origin_agent, url=args.url, domain=args.domain, source_ref=args.source_ref, evidence_ref=args.evidence_ref, metadata=metadata or None, limit=args.limit, ), indent=2, ) ) elif args.command == "task-bundle": print(json.dumps(service.task_bundle(args.task_id), indent=2)) elif args.command == "audit-v2": print(json.dumps(service.audit_v2(), indent=2)) elif args.command == "migrate-v2": print(json.dumps(service.migrate_v2(), indent=2)) elif args.command == "export": payload = service.export() if args.output: Path(args.output).write_text(json.dumps(payload, indent=2), encoding="utf-8") print(json.dumps({"output": str(Path(args.output).resolve())}, indent=2)) else: print(json.dumps(payload, indent=2)) if __name__ == "__main__": main()