| """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]) |
|
|
