File size: 3,277 Bytes
c3cec51 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 | """Rivet Context Loader — loads architecture + audit into every session.
The key innovation: Rivet starts every conversation KNOWING the system.
Not learning it from the user. Not guessing from file names. KNOWING.
"""
import json
from pathlib import Path
from dataclasses import dataclass
CONTEXT_DIR = Path("/Users/margaret/project-rivet/context")
@dataclass
class RivetContext:
architecture: str
audit_findings: str
schema_snapshot: str = ""
recent_git: str = ""
def to_system_context(self) -> str:
"""Format as a system context block for the model."""
parts = [
"# SYSTEM CONTEXT — Multiverse Campus Architecture\n",
"You have the following information loaded. Reference it when answering.\n",
"## Architecture Map\n",
self.architecture,
"\n## Known Issues (Nexus Security Audit, 2026-07-10)\n",
self.audit_findings,
]
if self.schema_snapshot:
parts.extend(["\n## Database Schema (key tables)\n", self.schema_snapshot])
if self.recent_git:
parts.extend(["\n## Recent Changes (last 7 days)\n", self.recent_git])
return "\n".join(parts)
@property
def token_estimate(self) -> int:
"""Rough token estimate (4 chars per token)."""
total = len(self.architecture) + len(self.audit_findings)
total += len(self.schema_snapshot) + len(self.recent_git)
return total // 4
def load_context(context_dir: Path = CONTEXT_DIR) -> RivetContext:
"""Load all context files from the Rivet context directory."""
arch_path = context_dir / "architecture_map.md"
audit_path = context_dir / "audit_findings.md"
schema_path = context_dir / "schema_snapshot.sql"
git_path = context_dir / "recent_git.txt"
architecture = arch_path.read_text() if arch_path.exists() else "Architecture map not loaded."
audit = audit_path.read_text() if audit_path.exists() else "Audit findings not loaded."
schema = schema_path.read_text() if schema_path.exists() else ""
git_log = git_path.read_text() if git_path.exists() else ""
return RivetContext(
architecture=architecture,
audit_findings=audit,
schema_snapshot=schema,
recent_git=git_log,
)
def refresh_git_log(repo_path: str, context_dir: Path = CONTEXT_DIR):
"""Refresh the recent git log from the campus repo."""
import subprocess
try:
result = subprocess.run(
["git", "log", "--oneline", "--since=7 days ago", "-20"],
cwd=repo_path, capture_output=True, text=True, timeout=10
)
if result.returncode == 0:
(context_dir / "recent_git.txt").write_text(result.stdout)
except (subprocess.TimeoutExpired, FileNotFoundError):
pass
if __name__ == "__main__":
ctx = load_context()
print(f"Context loaded:")
print(f" Architecture: {len(ctx.architecture)} chars")
print(f" Audit: {len(ctx.audit_findings)} chars")
print(f" Schema: {len(ctx.schema_snapshot)} chars")
print(f" Git: {len(ctx.recent_git)} chars")
print(f" Estimated tokens: ~{ctx.token_estimate}")
print(f"\nFirst 200 chars of system context:")
print(ctx.to_system_context()[:200])
|