my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	"""
#2	Clawd Brain - persistent Solana-native memory and wiki layer.
#3
#4	This module adapts the Mnemosyne engine into the Clawd Memory brain:
#5	- durable SQLite bank named "clawd"
#6	- Obsidian-compatible markdown vault with wiki links and frontmatter
#7	- Solana, perpetual trading, OODA, x402, A2A, and agent harness metadata
#8	- lightweight auto-research ingestion for URLs and research prompts
#9	"""
#10
#11	from __future__ import annotations
#12
#13	import argparse
#14	import hashlib
#15	import json
#16	import os
#17	import re
#18	import sqlite3
#19	import sys
#20	from dataclasses import dataclass, field
#21	from datetime import datetime, timezone
#22	from pathlib import Path
#23	from typing import Any, Dict, Iterable, List, Optional
#24	from urllib.error import URLError
#25	from urllib.parse import urlparse
#26	from urllib.request import Request, urlopen
#27
#28	from mnemosyne.core.banks import BankManager
#29	from mnemosyne.core.memory import Mnemosyne
#30
#31
#32	DEFAULT_BANK = "clawd"
#33	DEFAULT_SESSION = "clawd-brain"
#34	DEFAULT_VAULT_ENV = "CLAWD_BRAIN_VAULT"
#35	DEFAULT_DATA_ENV = "MNEMOSYNE_DATA_DIR"
#36	REPO_ROOT = Path(__file__).resolve().parents[2]
#37	DEFAULT_VAULT = REPO_ROOT / "MemeBRain" / "vault"
#38	OODA_JOURNAL = REPO_ROOT / "ooda" / "journal" / "ticks.jsonl"
#39
#40	DOMAIN_TAGS = {
#41	"solana",
#42	"clawd",
#43	"memecoin",
#44	"perp",
#45	"perpetual",
#46	"trading",
#47	"jupiter",
#48	"raydium",
#49	"orca",
#50	"meteora",
#51	"pumpfun",
#52	"helius",
#53	"metaplex",
#54	"x402",
#55	"a2a",
#56	"ap2",
#57	"mpp",
#58	"ooda",
#59	"risk",
#60	"agent",
#61	"wallet",
#62	"usdc",
#63	}
#64
#65
#66	def utc_now() -> str:
#67	return datetime.now(timezone.utc).replace(microsecond=0).isoformat()
#68
#69
#70	def slugify(value: str, fallback: str = "note") -> str:
#71	value = value.lower().strip()
#72	value = re.sub(r"[^a-z0-9]+", "-", value)
#73	value = re.sub(r"-+", "-", value).strip("-")
#74	return value[:96] or fallback
#75
#76
#77	def checksum(value: str) -> str:
#78	return hashlib.sha256(value.encode("utf-8")).hexdigest()[:16]
#79
#80
#81	def extract_wiki_links(content: str) -> List[str]:
#82	links = re.findall(r"\[\[([^\]\|]+)(?:\\|[^\]]+)?\]\]", content)
#83	return sorted({link.strip() for link in links if link.strip()})
#84
#85
#86	def detect_tags(content: str, extra: Iterable[str] = ()) -> List[str]:
#87	haystack = content.lower()
#88	tags = {tag for tag in DOMAIN_TAGS if tag in haystack}
#89	tags.update(t.strip().lower().replace("#", "") for t in extra if t.strip())
#90	return sorted(tags)
#91
#92
#93	def yaml_scalar(value: Any) -> str:
#94	if isinstance(value, bool):
#95	return "true" if value else "false"
#96	if isinstance(value, (int, float)):
#97	return str(value)
#98	text = str(value).replace('"', '\\"')
#99	return f'"{text}"'
#100
#101
#102	def frontmatter(metadata: Dict[str, Any]) -> str:
#103	lines = ["---"]
#104	for key in sorted(metadata):
#105	value = metadata[key]
#106	if value is None:
#107	continue
#108	if isinstance(value, list):
#109	lines.append(f"{key}:")
#110	for item in value:
#111	lines.append(f" - {yaml_scalar(item)}")
#112	else:
#113	lines.append(f"{key}: {yaml_scalar(value)}")
#114	lines.append("---")
#115	return "\n".join(lines)
#116
#117
#118	def strip_html(html: str) -> str:
#119	html = re.sub(r"(?is)<(script\|style).?>.?</\1>", " ", html)
#120	html = re.sub(r"(?s)<[^>]+>", " ", html)
#121	html = re.sub(r"\s+", " ", html)
#122	return html.strip()
#123
#124
#125	def html_title(html: str, fallback: str) -> str:
#126	match = re.search(r"(?is)<title[^>]>(.?)</title>", html)
#127	if not match:
#128	return fallback
#129	return re.sub(r"\s+", " ", strip_html(match.group(1))).strip() or fallback
#130
#131
#132	@dataclass
#133	class BrainConfig:
#134	bank: str = DEFAULT_BANK
#135	session_id: str = DEFAULT_SESSION
#136	vault_path: Path = field(
#137	default_factory=lambda: Path(os.environ.get(DEFAULT_VAULT_ENV, DEFAULT_VAULT))
#138	)
#139	author_id: str = "openclawd"
#140	author_type: str = "agent"
#141	channel_id: str = "solana-clawd"
#142
#143
#144	class ClawdBrain:
#145	"""Persistent Clawd memory system backed by Mnemosyne and markdown files."""
#146
#147	def __init__(self, config: Optional[BrainConfig] = None):
#148	self.config = config or BrainConfig()
#149	self.config.vault_path = Path(self.config.vault_path)
#150	self._ensure_layout()
#151	self.memory = self._open_memory()
#152	self._init_index_db()
#153
#154	def _ensure_layout(self) -> None:
#155	for dirname in (
#156	"00-inbox",
#157	"10-research",
#158	"20-signals",
#159	"30-trades",
#160	"40-agents",
#161	"50-protocols",
#162	"60-wallets",
#163	"70-perps",
#164	"90-indexes",
#165	):
#166	(self.config.vault_path / dirname).mkdir(parents=True, exist_ok=True)
#167
#168	def _open_memory(self) -> Mnemosyne:
#169	manager = BankManager()
#170	if self.config.bank != "default" and not manager.bank_exists(self.config.bank):
#171	manager.create_bank(self.config.bank)
#172	return Mnemosyne(
#173	session_id=self.config.session_id,
#174	bank=self.config.bank,
#175	author_id=self.config.author_id,
#176	author_type=self.config.author_type,
#177	channel_id=self.config.channel_id,
#178	)
#179
#180	@property
#181	def index_db_path(self) -> Path:
#182	return self.config.vault_path / "90-indexes" / "clawd-brain.db"
#183
#184	def _init_index_db(self) -> None:
#185	with sqlite3.connect(self.index_db_path) as conn:
#186	conn.execute(
#187	"""
#188	CREATE TABLE IF NOT EXISTS notes (
#189	id TEXT PRIMARY KEY,
#190	title TEXT NOT NULL,
#191	path TEXT NOT NULL UNIQUE,
#192	kind TEXT NOT NULL,
#193	source TEXT,
#194	tags_json TEXT NOT NULL,
#195	memory_id TEXT,
#196	created_at TEXT NOT NULL,
#197	updated_at TEXT NOT NULL
#198	)
#199	"""
#200	)
#201	conn.execute(
#202	"""
#203	CREATE TABLE IF NOT EXISTS links (
#204	source_id TEXT NOT NULL,
#205	target_title TEXT NOT NULL,
#206	created_at TEXT NOT NULL,
#207	PRIMARY KEY (source_id, target_title)
#208	)
#209	"""
#210	)
#211	conn.execute("CREATE INDEX IF NOT EXISTS idx_notes_kind ON notes(kind)")
#212	conn.execute("CREATE INDEX IF NOT EXISTS idx_notes_source ON notes(source)")
#213
#214	def _folder_for_kind(self, kind: str) -> str:
#215	return {
#216	"research": "10-research",
#217	"signal": "20-signals",
#218	"trade": "30-trades",
#219	"agent": "40-agents",
#220	"protocol": "50-protocols",
#221	"wallet": "60-wallets",
#222	"perp": "70-perps",
#223	}.get(kind, "00-inbox")
#224
#225	def remember(
#226	self,
#227	title: str,
#228	content: str,
#229	*,
#230	kind: str = "note",
#231	source: str = "clawd",
#232	tags: Iterable[str] = (),
#233	metadata: Optional[Dict[str, Any]] = None,
#234	importance: float = 0.65,
#235	) -> Dict[str, Any]:
#236	note_id = checksum(f"{kind}:{source}:{title}:{content}")
#237	created_at = utc_now()
#238	all_tags = detect_tags(f"{title}\n{content}", tags)
#239	note_metadata: Dict[str, Any] = {
#240	"id": note_id,
#241	"title": title,
#242	"kind": kind,
#243	"source": source,
#244	"created": created_at,
#245	"updated": created_at,
#246	"bank": self.config.bank,
#247	"chain": "solana",
#248	"tags": all_tags,
#249	}
#250	if metadata:
#251	note_metadata.update(metadata)
#252
#253	folder = self._folder_for_kind(kind)
#254	filename = f"{slugify(title)}-{note_id}.md"
#255	path = self.config.vault_path / folder / filename
#256	body = f"{frontmatter(note_metadata)}\n\n# {title}\n\n{content.strip()}\n"
#257	path.write_text(body, encoding="utf-8")
#258
#259	memory_text = (
#260	f"{title}\n\n{content.strip()}\n\n"
#261	f"Kind: {kind}. Source: {source}. Tags: {', '.join(all_tags)}. "
#262	f"Vault: {path.relative_to(self.config.vault_path)}."
#263	)
#264	memory_id = self.memory.remember(
#265	memory_text,
#266	source=f"clawd:{kind}:{source}",
#267	importance=importance,
#268	metadata={
#269	"note_id": note_id,
#270	"title": title,
#271	"kind": kind,
#272	"source": source,
#273	"tags": all_tags,
#274	"vault_path": str(path),
#275	**(metadata or {}),
#276	},
#277	scope="global",
#278	extract_entities=True,
#279	)
#280
#281	links = extract_wiki_links(content)
#282	with sqlite3.connect(self.index_db_path) as conn:
#283	conn.execute(
#284	"""
#285	INSERT OR REPLACE INTO notes
#286	(id, title, path, kind, source, tags_json, memory_id, created_at, updated_at)
#287	VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
#288	""",
#289	(
#290	note_id,
#291	title,
#292	str(path),
#293	kind,
#294	source,
#295	json.dumps(all_tags),
#296	memory_id,
#297	created_at,
#298	created_at,
#299	),
#300	)
#301	for link in links:
#302	conn.execute(
#303	"INSERT OR IGNORE INTO links (source_id, target_title, created_at) VALUES (?, ?, ?)",
#304	(note_id, link, created_at),
#305	)
#306
#307	return {
#308	"id": note_id,
#309	"memory_id": memory_id,
#310	"path": str(path),
#311	"tags": all_tags,
#312	"links": links,
#313	}
#314
#315	def recall(self, query: str, top_k: int = 8) -> Dict[str, Any]:
#316	memories = self.memory.recall(query, top_k=top_k)
#317	notes = self._search_notes(query, top_k=top_k)
#318	return {"query": query, "memories": memories, "notes": notes}
#319
#320	def _search_notes(self, query: str, top_k: int = 8) -> List[Dict[str, Any]]:
#321	terms = [t for t in re.split(r"\W+", query.lower()) if len(t) > 2]
#322	if not terms:
#323	return []
#324	results: List[Dict[str, Any]] = []
#325	with sqlite3.connect(self.index_db_path) as conn:
#326	conn.row_factory = sqlite3.Row
#327	rows = conn.execute("SELECT * FROM notes ORDER BY updated_at DESC").fetchall()
#328	for row in rows:
#329	text = f"{row['title']} {row['kind']} {row['source']} {row['tags_json']}".lower()
#330	score = sum(1 for term in terms if term in text)
#331	if score:
#332	item = dict(row)
#333	item["score"] = score
#334	item["tags"] = json.loads(item.pop("tags_json"))
#335	results.append(item)
#336	return sorted(results, key=lambda item: item["score"], reverse=True)[:top_k]
#337
#338	def auto_research(self, target: str, *, tags: Iterable[str] = ()) -> Dict[str, Any]:
#339	"""Archive a URL or create a research task note for a topic."""
#340	if target.startswith(("http://", "https://")):
#341	title, content = self._fetch_url(target)
#342	return self.remember(
#343	title,
#344	content,
#345	kind="research",
#346	source=target,
#347	tags=[*tags, "auto-research"],
#348	metadata={"url": target, "research_status": "archived"},
#349	importance=0.72,
#350	)
#351
#352	prompt = (
#353	f"Research target: [[{target}]]\n\n"
#354	"Checklist:\n"
#355	"- Solana token, wallet, protocol, or perp venue relevance\n"
#356	"- On-chain evidence to verify with Helius or RPC tools\n"
#357	"- Liquidity, holder, volume, unlock, and routing risks\n"
#358	"- Trading thesis, invalidation, and risk controls\n"
#359	"- Follow-up URLs and data sources\n"
#360	)
#361	return self.remember(
#362	f"Research Queue - {target}",
#363	prompt,
#364	kind="research",
#365	source="auto-research",
#366	tags=[*tags, "research-queue"],
#367	metadata={"research_status": "queued"},
#368	importance=0.55,
#369	)
#370
#371	def _fetch_url(self, url: str) -> tuple[str, str]:
#372	req = Request(url, headers={"User-Agent": "ClawdMemory/1.0"})
#373	try:
#374	with urlopen(req, timeout=15) as response:
#375	raw = response.read(750_000)
#376	content_type = response.headers.get("content-type", "")
#377	except URLError as exc:
#378	raise RuntimeError(f"research fetch failed for {url}: {exc}") from exc
#379
#380	text = raw.decode("utf-8", errors="replace")
#381	parsed = urlparse(url)
#382	fallback = parsed.netloc + parsed.path
#383	if "html" in content_type.lower() or "<html" in text[:500].lower():
#384	title = html_title(text, fallback)
#385	content = strip_html(text)[:25_000]
#386	else:
#387	title = fallback
#388	content = text[:25_000]
#389	return title, f"Source URL: {url}\n\n{content}"
#390
#391	def ingest_ooda_journal(self, journal_path: Path = OODA_JOURNAL, limit: int = 100) -> Dict[str, Any]:
#392	journal_path = Path(journal_path)
#393	if not journal_path.exists():
#394	raise FileNotFoundError(f"OODA journal not found: {journal_path}")
#395	lines = [line for line in journal_path.read_text(encoding="utf-8").splitlines() if line.strip()]
#396	imported = 0
#397	for line in lines[-limit:]:
#398	entry = json.loads(line)
#399	tick = entry.get("tick", imported + 1)
#400	decision = entry.get("decision", {})
#401	title = f"OODA Tick {tick} - {decision.get('action', 'unknown')}"
#402	content = (
#403	"```json\n"
#404	f"{json.dumps(entry, indent=2, sort_keys=True)}\n"
#405	"```\n\n"
#406	f"Decision: {decision.get('action', 'unknown')} "
#407	f"{decision.get('side', '')}. Outcome: {entry.get('outcome')}."
#408	)
#409	self.remember(
#410	title,
#411	content,
#412	kind="signal",
#413	source="ooda-journal",
#414	tags=["ooda", "trading", "paper"],
#415	metadata={"tick": tick, "outcome": entry.get("outcome")},
#416	importance=0.6,
#417	)
#418	imported += 1
#419	return {"journal": str(journal_path), "imported": imported, "bank": self.config.bank}
#420
#421	def status(self) -> Dict[str, Any]:
#422	with sqlite3.connect(self.index_db_path) as conn:
#423	conn.row_factory = sqlite3.Row
#424	note_count = conn.execute("SELECT COUNT(*) AS c FROM notes").fetchone()["c"]
#425	by_kind = {
#426	row["kind"]: row["c"]
#427	for row in conn.execute("SELECT kind, COUNT(*) AS c FROM notes GROUP BY kind")
#428	}
#429	link_count = conn.execute("SELECT COUNT(*) AS c FROM links").fetchone()["c"]
#430	stats = self.memory.get_stats()
#431	return {
#432	"bank": self.config.bank,
#433	"vault": str(self.config.vault_path),
#434	"index_db": str(self.index_db_path),
#435	"notes": note_count,
#436	"links": link_count,
#437	"by_kind": by_kind,
#438	"memory": stats,
#439	}
#440
#441
#442	def _print_json(value: Any) -> None:
#443	print(json.dumps(value, indent=2, default=str))
#444
#445
#446	def build_parser() -> argparse.ArgumentParser:
#447	parser = argparse.ArgumentParser(description="Clawd Memory persistent brain and wiki")
#448	parser.add_argument("--vault", default=None, help="Vault path. Defaults to CLAWD_BRAIN_VAULT or MemeBRain/vault.")
#449	parser.add_argument("--bank", default=DEFAULT_BANK, help="Memory bank. Defaults to clawd.")
#450	sub = parser.add_subparsers(dest="command", required=True)
#451
#452	sub.add_parser("init", help="Create the vault layout and memory bank")
#453
#454	remember = sub.add_parser("remember", help="Write a wiki note and memory")
#455	remember.add_argument("title")
#456	remember.add_argument("content")
#457	remember.add_argument("--kind", default="note")
#458	remember.add_argument("--source", default="clawd")
#459	remember.add_argument("--tag", action="append", default=[])
#460	remember.add_argument("--importance", type=float, default=0.65)
#461
#462	recall = sub.add_parser("recall", help="Recall memories and matching vault notes")
#463	recall.add_argument("query")
#464	recall.add_argument("--top-k", type=int, default=8)
#465
#466	research = sub.add_parser("research", help="Archive URL or queue topic research")
#467	research.add_argument("target")
#468	research.add_argument("--tag", action="append", default=[])
#469
#470	ingest = sub.add_parser("ingest-ooda", help="Import OODA ticks into memory")
#471	ingest.add_argument("--journal", default=str(OODA_JOURNAL))
#472	ingest.add_argument("--limit", type=int, default=100)
#473
#474	sub.add_parser("status", help="Show brain status")
#475	return parser
#476
#477
#478	def run_cli(argv: Optional[List[str]] = None) -> None:
#479	args = build_parser().parse_args(argv)
#480	config = BrainConfig(bank=args.bank)
#481	if args.vault:
#482	config.vault_path = Path(args.vault)
#483	brain = ClawdBrain(config)
#484
#485	if args.command == "init":
#486	_print_json(brain.status())
#487	elif args.command == "remember":
#488	_print_json(
#489	brain.remember(
#490	args.title,
#491	args.content,
#492	kind=args.kind,
#493	source=args.source,
#494	tags=args.tag,
#495	importance=args.importance,
#496	)
#497	)
#498	elif args.command == "recall":
#499	_print_json(brain.recall(args.query, top_k=args.top_k))
#500	elif args.command == "research":
#501	_print_json(brain.auto_research(args.target, tags=args.tag))
#502	elif args.command == "ingest-ooda":
#503	_print_json(brain.ingest_ooda_journal(Path(args.journal), limit=args.limit))
#504	elif args.command == "status":
#505	_print_json(brain.status())
#506	else:
#507	raise SystemExit(f"unknown command: {args.command}")
#508
#509
#510	if __name__ == "__main__":
#511	try:
#512	run_cli()
#513	except Exception as exc:
#514	print(f"Error: {exc}", file=sys.stderr)
#515	raise SystemExit(1)
#516

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public