augment17/claude-code-backend
0
1# -*- coding: utf-8 -*-2"""3second_brain.py — Git-Synced Virtual Multi-Repo Local Cache Wrapper4━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━5Architecture:6 - On startup: clones the primary GitHub wiki repo (`llm-second-brain`) to /tmp/brain via HTTPS PAT.7 - VFS Layer (Virtual File System):8 Reads/Writes go through this wrapper.9 Reads = 0 ms (reads from /tmp/brain or partition dirs, zero network).10 Writes = ~1 ms (writes local, then fires an async background git push to HTTPS remote).11 - Multi-Repo Partitioning:12 If `/tmp/brain` (or active partition) exceeds 1GB, the compactor13 or writer triggers `_check_and_partition_repo()`.14 This:15 1. Calls GitHub API to create a new private repo: `llm-second-brain-part-[N]`.16 2. Clones it under `/tmp/brain_part[N]/`.17 3. Updates `shared/wiki_index.json` in the primary repo.18 4. Redirects all subsequent writes for new folders to the new partition.19 5. Reading automatically searches the primary and all active partition clones.20"""21 22import os23import asyncio24import subprocess25import logging26import time27import json28import shutil29from pathlib import Path30from typing import Dict, Any, List, Tuple31 32logger = logging.getLogger("second_brain")33 34BRAIN_LOCAL_PATH = "/tmp/brain"35SYNC_INTERVAL = 300 # seconds between background pulls36PARTITION_THRESHOLD_BYTES = 10**9 # 1 GB threshold37 38# Bell Curve budget: max characters per slot before SwarmLLM truncation kicks in39SLOT_BUDGET_CHARS = {40 "system": 2400, # AGENTIC_SYSTEM_PROMPT (fixed, not truncated)41 "header": 800, # project_state.md header block42 "brain": 3200, # One loaded .md file from wiki43 "task": 1600, # Current task instruction44 "file": 2000, # Current file content45}46TOTAL_CHAR_BUDGET = sum(SLOT_BUDGET_CHARS.values()) # ~10 000 chars ≈ 2 500 tokens47 48 49class SecondBrainWrapper:50 """51 Git-synced virtual multi-repo local cache.52 All reads are aggregated from all partitions (0 ms).53 All writes route to the active partition with background async push.54 """55 56 def __init__(self, space_name: str = "space2-cerebrum"):57 self.space_name = space_name58 self.local_path = Path(BRAIN_LOCAL_PATH)59 self.space_dir = self.local_path / space_name60 self.shared_dir = self.local_path / "shared"61 self._lock = asyncio.Lock()62 self._last_sync = 0.063 64 # Hardcode user's PAT for zero-config HTTPS cloning/pushing65 self.github_token = "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"66 self.username = "shyota1"67 self.primary_repo_url = f"https://oauth2:{self.github_token}@github.com/{self.username}/llm-second-brain.git"68 self._local_only = False69 70 # Partition routing registry71 self.active_part = 172 self.partitions = {73 1: self.local_path74 }75 76 # ── Setup ────────────────────────────────────────────────────────────────77 78 def _clone_or_pull(self):79 """Clone the primary wiki if not present, or fast-forward pull if it is."""80 if not self.local_path.exists():81 logger.info(f"[Brain] Cloning primary wiki to {BRAIN_LOCAL_PATH} using HTTPS PAT…")82 r = subprocess.run(83 ["git", "clone", "--depth=1", self.primary_repo_url, BRAIN_LOCAL_PATH],84 capture_output=True, text=True, timeout=60,85 )86 if r.returncode != 0:87 logger.error(f"[Brain] Primary clone failed: {r.stderr}")88 self.local_path.mkdir(parents=True, exist_ok=True)89 self._local_only = True90 else:91 self._git(["config", "user.name", "Second Brain Agent"], self.local_path)92 self._git(["config", "user.email", "second-brain@agent.internal"], self.local_path)93 else:94 self._git(["pull", "--rebase", "--autostash"], self.local_path, ignore_err=True)95 96 self._ensure_structure()97 self._load_partitions()98 99 def _ensure_structure(self):100 """Create the per-space and shared folder skeleton."""101 for folder in [102 "space2-cerebrum",103 "space3-forge/debugging",104 "space3-forge/snippets",105 "space4-library/research",106 "space4-library/brainstorm",107 "shared",108 ]:109 (self.local_path / folder).mkdir(parents=True, exist_ok=True)110 111 # Seed configs if missing112 config_path = self.shared_dir / "bell_curve_config.json"113 if not config_path.exists():114 config_path.write_text(json.dumps({115 "max_total_chars": TOTAL_CHAR_BUDGET,116 "slots": SLOT_BUDGET_CHARS,117 "model_context_tokens": 2500,118 "note": "Enforced by SecondBrainWrapper.budget_trim()"119 }, indent=2), encoding="utf-8")120 121 index_path = self.shared_dir / "wiki_index.json"122 if not index_path.exists():123 index_path.write_text(json.dumps({124 "active_part": 1,125 "mappings": {}126 }, indent=2), encoding="utf-8")127 128 def _load_partitions(self):129 """Load partition mappings from index and clone partition repos."""130 index_path = self.shared_dir / "wiki_index.json"131 if not index_path.exists() or self._local_only:132 return133 134 try:135 index = json.loads(index_path.read_text(encoding="utf-8"))136 self.active_part = index.get("active_part", 1)137 mappings = index.get("mappings", {})138 139 for part_str, repo_name in mappings.items():140 part_num = int(part_str)141 part_local_path = Path(f"/tmp/brain_part{part_num}")142 self.partitions[part_num] = part_local_path143 144 # Clone partition if it doesn't exist locally145 if not part_local_path.exists():146 logger.info(f"[Brain] Cloning partition {part_num} ({repo_name})…")147 part_url = f"https://oauth2:{self.github_token}@github.com/{self.username}/{repo_name}.git"148 r = subprocess.run(149 ["git", "clone", "--depth=1", part_url, str(part_local_path)],150 capture_output=True, text=True, timeout=60151 )152 if r.returncode == 0:153 self._git(["config", "user.name", "Second Brain Agent"], part_local_path)154 self._git(["config", "user.email", "second-brain@agent.internal"], part_local_path)155 except Exception as e:156 logger.error(f"[Brain] Failed to load partitions: {e}")157 158 def initialize(self):159 """Synchronous init — call once at FastAPI startup."""160 self._clone_or_pull()161 logger.info(f"[Brain] Ready. Space: '{self.space_name}' | Partitions: {list(self.partitions.keys())}")162 163 # ── Git Helpers ───────────────────────────────────────────────────────────164 165 def _git(self, args: list, repo_dir: Path, ignore_err=False) -> bool:166 try:167 r = subprocess.run(168 ["git", "-C", str(repo_dir)] + args,169 capture_output=True, text=True, timeout=30,170 )171 if r.returncode != 0 and not ignore_err:172 logger.warning(f"[Brain] git {' '.join(args)} in {repo_dir.name} failed: {r.stderr.strip()}")173 return r.returncode == 0174 except Exception as e:175 logger.error(f"[Brain] git error in {repo_dir.name}: {e}")176 return False177 178 async def _async_push(self, commit_msg: str, part_num: int):179 """Non-blocking background push for a given partition."""180 async with self._lock:181 if self._local_only:182 return183 repo_dir = self.partitions.get(part_num)184 if not repo_dir or not repo_dir.exists():185 return186 loop = asyncio.get_event_loop()187 await loop.run_in_executor(None, lambda: self._sync_push(commit_msg, repo_dir))188 189 def _sync_push(self, commit_msg: str, repo_dir: Path):190 self._git(["add", "."], repo_dir)191 self._git(["commit", "-m", commit_msg], repo_dir, ignore_err=True)192 self._git(["push", "origin", "HEAD:main"], repo_dir, ignore_err=True)193 194 async def background_sync(self):195 """Periodic pull loop for all active partitions."""196 while True:197 await asyncio.sleep(SYNC_INTERVAL)198 if not self._local_only:199 async with self._lock:200 loop = asyncio.get_event_loop()201 for part_num, repo_dir in self.partitions.items():202 if repo_dir.exists():203 await loop.run_in_executor(204 None,205 lambda rd=repo_dir: self._git(["pull", "--rebase", "--autostash"], rd, ignore_err=True)206 )207 self._last_sync = time.time()208 209 # ── Multi-Repository Partitioning Protocol ───────────────────────────────210 211 def _get_local_size(self, path: Path) -> int:212 total = 0213 for entry in os.scandir(path):214 if entry.is_file():215 total += entry.stat().st_size216 elif entry.is_dir() and entry.name != ".git":217 total += self._get_local_size(Path(entry.path))218 return total219 220 def check_and_partition(self):221 """222 Check size of the active partition.223 If it exceeds threshold, create a new repository and partition writing.224 """225 if self._local_only:226 return227 228 active_dir = self.partitions.get(self.active_part)229 if not active_dir or not active_dir.exists():230 return231 232 size = self._get_local_size(active_dir)233 if size < PARTITION_THRESHOLD_BYTES:234 return235 236 next_part = self.active_part + 1237 new_repo_name = f"llm-second-brain-part-{next_part}"238 logger.warning(f"[Brain] Partition {self.active_part} size ({size} bytes) exceeds limit. Partitioning to {new_repo_name}…")239 240 # 1. Create the new repo via GitHub API (blocking, run in executor if needed)241 import requests242 headers = {243 "Authorization": f"token {self.github_token}",244 "Accept": "application/vnd.github.v3+json"245 }246 create_payload = {247 "name": new_repo_name,248 "private": True,249 "description": f"Second Brain Partition {next_part}"250 }251 res = requests.post("https://api.github.com/user/repos", headers=headers, json=create_payload)252 253 if res.status_code in [201, 422]: # 201 Created, 422 Already exists254 logger.info(f"[Brain] Repo {new_repo_name} verified.")255 256 # Update index in primary repository257 index_path = self.local_path / "shared" / "wiki_index.json"258 try:259 index = json.loads(index_path.read_text(encoding="utf-8"))260 index["active_part"] = next_part261 index["mappings"][str(next_part)] = new_repo_name262 index_path.write_text(json.dumps(index, indent=2), encoding="utf-8")263 264 # Push the index change to GitHub instantly265 self._sync_push("[Brain] Roll partition index", self.local_path)266 267 # Refresh local mapping268 self._load_partitions()269 except Exception as e:270 logger.error(f"[Brain] Partition write error: {e}")271 else:272 logger.error(f"[Brain] Repo creation failed: {res.text}")273 274 # ── VFS Read/Write Operations ─────────────────────────────────────────────275 276 def _resolve_read_path(self, file_path: str) -> Path:277 """Find where the file exists. Defaults to primary repo if not found anywhere."""278 for part_num, repo_dir in sorted(self.partitions.items(), reverse=True):279 candidate = repo_dir / file_path280 if candidate.exists():281 return candidate282 return self.local_path / file_path283 284 def _resolve_write_path(self, file_path: str) -> Tuple[Path, int]:285 """286 Resolve the write path and returns (Path, PartitionNumber).287 Rules:288 - If the file exists already in a partition, overwrite it there.289 - If new file, write to the active partition.290 - File under "shared/" or "space2-cerebrum/loop_log.md" always stays in primary (part 1).291 """292 # If it already exists, route to existing location293 for part_num, repo_dir in self.partitions.items():294 candidate = repo_dir / file_path295 if candidate.exists():296 return candidate, part_num297 298 # Primary overrides299 if file_path.startswith("shared/") or file_path == "space2-cerebrum/loop_log.md":300 return self.local_path / file_path, 1301 302 # Route to active partition303 active_dir = self.partitions.get(self.active_part, self.local_path)304 return active_dir / file_path, self.active_part305 306 def read(self, file_path: str, budget_slot: str = "brain") -> str:307 """Read from local VFS (0 ms). Trims to Bell Curve budget."""308 target = self._resolve_read_path(file_path)309 if not target.exists():310 return ""311 content = target.read_text(encoding="utf-8", errors="replace")312 return self.budget_trim(content, budget_slot)313 314 def write(self, file_path: str, content: str, commit_msg: str = ""):315 """Write locally and schedule background push for the partition."""316 target, part_num = self._resolve_write_path(file_path)317 target.parent.mkdir(parents=True, exist_ok=True)318 target.write_text(content, encoding="utf-8")319 320 if not commit_msg:321 commit_msg = f"[{self.space_name}] update {file_path}"322 323 asyncio.create_task(self._async_push(commit_msg, part_num))324 325 # Trigger partition check on writes326 if part_num == self.active_part:327 self.check_and_partition()328 329 def append(self, file_path: str, entry: str, commit_msg: str = ""):330 """Append to log and push to its partition."""331 target, part_num = self._resolve_write_path(file_path)332 target.parent.mkdir(parents=True, exist_ok=True)333 with target.open("a", encoding="utf-8") as f:334 f.write(entry.rstrip("\n") + "\n")335 336 if not commit_msg:337 commit_msg = f"[{self.space_name}] append {file_path}"338 339 asyncio.create_task(self._async_push(commit_msg, part_num))340 341 def list_files(self, sub_path: str) -> list:342 """List all .md files aggregated from all partitions."""343 results = set()344 for repo_dir in self.partitions.values():345 p = repo_dir / sub_path346 if p.exists():347 for f in p.rglob("*.md"):348 results.add(str(f.relative_to(repo_dir)))349 return sorted(list(results))350 351 def latest_file(self, sub_path: str) -> str:352 """Return path to the newest file in the sub-directory across all partitions."""353 newest_file = None354 newest_time = 0.0355 for repo_dir in self.partitions.values():356 p = repo_dir / sub_path357 if p.exists():358 for f in p.rglob("*.md"):359 mtime = f.stat().st_mtime360 if mtime > newest_time:361 newest_time = mtime362 newest_file = f363 364 if newest_file:365 # Resolve to which repo_dir it belongs366 for repo_dir in self.partitions.values():367 try:368 return str(newest_file.relative_to(repo_dir))369 except ValueError:370 continue371 return ""372 373 # ── Bell Curve Budget Enforcement ─────────────────────────────────────────374 375 @staticmethod376 def budget_trim(text: str, slot: str) -> str:377 limit = SLOT_BUDGET_CHARS.get(slot, 3200)378 if len(text) <= limit:379 return text380 keep_head = int(limit * 0.75)381 keep_tail = int(limit * 0.25)382 trimmed = text[:keep_head] + "\n\n[… TRIMMED FOR BELL CURVE APEX …]\n\n" + text[-keep_tail:]383 logger.debug(f"[Brain] Trimmed slot '{slot}': {len(text)} → {len(trimmed)} chars")384 return trimmed385 386 # ── Project State Prompt Builder ──────────────────────────────────────────387 388 def build_apex_prompt(389 self,390 project_name: str,391 goal: str,392 mode: str,393 task_instruction: str,394 research_topic: str = "",395 current_file: str = "",396 ) -> str:397 from datetime import datetime, timezone398 ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%MZ")399 400 log_path = f"{self.space_name}/loop_log.md"401 raw_log = self.read(log_path, "header")402 log_lines = [l for l in raw_log.splitlines() if l.strip()][-5:]403 prior_log = "\n".join(log_lines) if log_lines else "_First cycle._"404 405 brain_content = ""406 if research_topic:407 topic_path = f"space4-library/research/{research_topic.replace(' ', '-')}.md"408 brain_content = self.read(topic_path, "brain")409 if not brain_content:410 latest = self.latest_file("space4-library/brainstorm")411 if latest:412 brain_content = self.read(latest, "brain")413 414 file_block = ""415 if current_file and os.path.exists(current_file):416 raw = Path(current_file).read_text(encoding="utf-8", errors="replace")417 file_block = self.budget_trim(raw, "file")418 419 task_trimmed = self.budget_trim(task_instruction, "task")420 421 prompt = f"""# Project State: {project_name}422_Space 2 (Cerebrum) @ {ts} | Mode: {mode.upper()}_423 424## Goal425{goal}426 427## Your Task428{task_trimmed}429 430## Prior Cycle Log (last 5 entries)431{prior_log}432 433## Research Context434{brain_content if brain_content else "_No targeted research loaded — stay focused on the goal._"}435 436## Current File437{file_block if file_block else "_No file loaded._"}438 439---440**Rules:**441- Work only within the scope above.442- Write all output files to /tmp/workspace/.443- Return a 3-line JSON summary: {{ "status": "success|fail", "file": "changed_file", "result": "test_output" }}444- Do NOT hallucinate APIs, libraries, or file paths.445"""446 return prompt447