Initial v0.1: memwalk CLI on top of memba
- Config in ~/.memwalk/config.toml (typer init writes it) - Sources: git log walker + bash history parser with secret/noise filters - Ingest orchestrator feeds events into a memba Session and saves ~/.memwalk/current.memb via the rewritten Session.chat() that preserves loaded state (raw eval+sample, no KV-cache reset) - Daily snapshot rotation under ~/.memwalk/snapshots/ - CLI commands: init, update, ask, standup, status, prune Validated on Nemotron-3-Nano-4B Q4_K_M end-to-end: ingest 304 events (303 commits + 1 shell session) in 2.8s on GPU, recall in fresh process returns accurate per-project summaries. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
+26
@@ -0,0 +1,26 @@
|
||||
# Python
|
||||
__pycache__/
|
||||
*.py[cod]
|
||||
*.egg-info/
|
||||
.pytest_cache/
|
||||
.mypy_cache/
|
||||
.ruff_cache/
|
||||
.venv/
|
||||
venv/
|
||||
|
||||
# Build
|
||||
build/
|
||||
dist/
|
||||
|
||||
# IDE
|
||||
.vscode/
|
||||
.idea/
|
||||
*.swp
|
||||
.DS_Store
|
||||
|
||||
# Claude Code
|
||||
.claude/
|
||||
|
||||
# Local state files (the whole point is they're per-user)
|
||||
*.memb
|
||||
.memwalk/
|
||||
@@ -0,0 +1,69 @@
|
||||
# memwalk
|
||||
|
||||
> Walk through your work memory.
|
||||
|
||||
A local-first CLI that watches your git activity (and optionally your shell
|
||||
history) and feeds it into a Mamba-based LLM via persistent state. You can
|
||||
then ask in plain English what you were doing last week, why you started that
|
||||
branch, or generate a standup from yesterday's commits — without your data
|
||||
ever leaving the machine.
|
||||
|
||||
Built on **[memba](https://github.com/emil28092005/Memba)** for state
|
||||
persistence and **[NVIDIA Nemotron-3-Nano-4B](https://huggingface.co/nvidia/NVIDIA-Nemotron-3-Nano-4B-GGUF)**
|
||||
(or any other GGUF SSM/hybrid model) for inference.
|
||||
|
||||
## Status
|
||||
|
||||
v0.1 — alpha. Works end-to-end on Linux for the maintainer; APIs and on-disk
|
||||
format may change.
|
||||
|
||||
## Install
|
||||
|
||||
```bash
|
||||
# Until memba is on PyPI, install both editable from local clones:
|
||||
pip install -e /path/to/Memba
|
||||
pip install -e /path/to/memwalk
|
||||
```
|
||||
|
||||
Make sure you have a GGUF Mamba-2 or hybrid model. Recommended:
|
||||
|
||||
```bash
|
||||
hf download nvidia/NVIDIA-Nemotron-3-Nano-4B-GGUF \
|
||||
NVIDIA-Nemotron3-Nano-4B-Q4_K_M.gguf \
|
||||
--local-dir ~/.memwalk/models
|
||||
```
|
||||
|
||||
## Quickstart
|
||||
|
||||
```bash
|
||||
memwalk init # interactive setup
|
||||
memwalk update # ingest last 30 days of git+bash
|
||||
memwalk standup # auto-generate daily standup
|
||||
memwalk ask "What was I working on last week?"
|
||||
memwalk status
|
||||
```
|
||||
|
||||
## What it actually does
|
||||
|
||||
`memwalk update` walks your configured git repos and (optionally) your bash
|
||||
history, formats new events into a readable activity block, and feeds that
|
||||
into the SSM model. The model's hidden state — a fixed ~85 MB blob — is
|
||||
saved to `~/.memwalk/current.memb` via memba.
|
||||
|
||||
`memwalk ask` and `memwalk standup` load that state and query it. The model
|
||||
recalls themes, projects, and trajectory across processes and reboots.
|
||||
|
||||
## Layout
|
||||
|
||||
```
|
||||
~/.memwalk/
|
||||
├── config.toml
|
||||
├── current.memb ← rolling state
|
||||
├── last_update.txt
|
||||
└── snapshots/
|
||||
└── 2026-05-16.memb ← daily snapshot before each update
|
||||
```
|
||||
|
||||
## License
|
||||
|
||||
MIT.
|
||||
@@ -0,0 +1,3 @@
|
||||
"""memwalk — semantic recall of your work via local SSM models."""
|
||||
|
||||
__version__ = "0.1.0"
|
||||
+187
@@ -0,0 +1,187 @@
|
||||
"""memwalk CLI — typer entry point."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timedelta
|
||||
from pathlib import Path
|
||||
|
||||
import typer
|
||||
from rich.console import Console
|
||||
from rich.prompt import Prompt
|
||||
from rich.table import Table
|
||||
|
||||
from . import __version__
|
||||
from .config import (
|
||||
DEFAULT_MODEL_HINT, HOME_DIR, BashConfig, Config, GitConfig,
|
||||
load_config, read_last_update, write_config,
|
||||
)
|
||||
from .ingest import open_session, query, update
|
||||
from .snapshot import prune_old
|
||||
from .sources import bash as bash_src
|
||||
from .sources import git as git_src
|
||||
|
||||
cli = typer.Typer(name="memwalk", help="Walk through your work memory.",
|
||||
add_completion=False)
|
||||
console = Console()
|
||||
|
||||
|
||||
# ── init ──────────────────────────────────────────────────────────
|
||||
|
||||
@cli.command()
|
||||
def init(
|
||||
model: str = typer.Option(None, "--model", help="Path to GGUF model"),
|
||||
scan_path: list[str] = typer.Option(None, "--scan",
|
||||
help="Directory to scan for git repos (repeatable)"),
|
||||
no_bash: bool = typer.Option(False, "--no-bash", help="Disable bash history"),
|
||||
force: bool = typer.Option(False, "--force", "-f",
|
||||
help="Overwrite existing config"),
|
||||
) -> None:
|
||||
"""Interactive (or flag-driven) setup. Writes ~/.memwalk/config.toml."""
|
||||
config_path = HOME_DIR / "config.toml"
|
||||
if config_path.exists() and not force:
|
||||
console.print(f"[yellow]Config already exists at {config_path}[/yellow]")
|
||||
console.print("Use --force to overwrite, or edit the file by hand.")
|
||||
raise typer.Exit(1)
|
||||
|
||||
console.print(f"[bold cyan]memwalk init v{__version__}[/bold cyan]\n")
|
||||
|
||||
# Model
|
||||
if model is None:
|
||||
console.print(DEFAULT_MODEL_HINT + "\n")
|
||||
model = Prompt.ask("Path to GGUF model")
|
||||
model_p = Path(model).expanduser()
|
||||
if not model_p.exists():
|
||||
console.print(f"[yellow]warning: {model_p} doesn't exist yet[/yellow]")
|
||||
|
||||
# Scan paths
|
||||
if scan_path:
|
||||
scan_paths = [Path(p).expanduser() for p in scan_path]
|
||||
else:
|
||||
default = str(Path.home() / "Desktop/Coding")
|
||||
raw = Prompt.ask(
|
||||
"Directories to scan for git repos (comma-separated)",
|
||||
default=default,
|
||||
)
|
||||
scan_paths = [Path(p.strip()).expanduser() for p in raw.split(",") if p.strip()]
|
||||
|
||||
cfg = Config(
|
||||
model_path=model_p,
|
||||
git=GitConfig(scan_paths=scan_paths),
|
||||
bash=BashConfig(enabled=not no_bash),
|
||||
)
|
||||
write_config(cfg)
|
||||
|
||||
console.print(f"\n[green]✓[/green] Wrote {cfg.config_path}")
|
||||
console.print(f"[green]✓[/green] State dir: {cfg.state_dir}")
|
||||
console.print(f"\nNext: [bold]memwalk update[/bold] to ingest the last 30 days")
|
||||
|
||||
|
||||
# ── update ────────────────────────────────────────────────────────
|
||||
|
||||
@cli.command(name="update")
|
||||
def update_cmd(
|
||||
verbose: bool = typer.Option(False, "--verbose", "-v"),
|
||||
) -> None:
|
||||
"""Ingest new git+bash events since last update."""
|
||||
cfg = load_config()
|
||||
with console.status("Ingesting…"):
|
||||
result = update(cfg, verbose=verbose)
|
||||
|
||||
if result["ingested"] == 0:
|
||||
console.print(f"[dim]No new activity since {result['since'].strftime('%Y-%m-%d %H:%M')}[/dim]")
|
||||
return
|
||||
|
||||
console.print(
|
||||
f"[green]✓[/green] Ingested {result['ingested']} events "
|
||||
f"([cyan]{result['git']}[/cyan] commits + "
|
||||
f"[cyan]{result['bash']}[/cyan] shell sessions) "
|
||||
f"in {result['elapsed_s']:.1f}s"
|
||||
)
|
||||
console.print(f" Window: {result['since'].strftime('%Y-%m-%d %H:%M')} → "
|
||||
f"{result['until'].strftime('%Y-%m-%d %H:%M')}")
|
||||
console.print(f" State : {result['state_size']:,} bytes"
|
||||
+ (" (daily snapshot taken)" if result["snapshotted"] else ""))
|
||||
|
||||
|
||||
# ── ask ───────────────────────────────────────────────────────────
|
||||
|
||||
@cli.command()
|
||||
def ask(
|
||||
question: str = typer.Argument(..., help="What to ask the model"),
|
||||
max_tokens: int = typer.Option(400, "--max-tokens"),
|
||||
verbose: bool = typer.Option(False, "--verbose", "-v"),
|
||||
) -> None:
|
||||
"""Query the current state."""
|
||||
cfg = load_config()
|
||||
with console.status("Thinking…"):
|
||||
answer = query(cfg, question, max_tokens=max_tokens, verbose=verbose)
|
||||
console.print(answer)
|
||||
|
||||
|
||||
# ── standup ───────────────────────────────────────────────────────
|
||||
|
||||
@cli.command()
|
||||
def standup(
|
||||
verbose: bool = typer.Option(False, "--verbose", "-v"),
|
||||
) -> None:
|
||||
"""Generate a brief 'what I did + what's next' summary from recent activity."""
|
||||
cfg = load_config()
|
||||
q = (
|
||||
"Generate my daily standup notes. Cover: what I worked on yesterday "
|
||||
"(grouped by project), what I plan today based on the trajectory, and "
|
||||
"any blockers visible in the activity. Be concise — bullet points, "
|
||||
"no preamble."
|
||||
)
|
||||
with console.status("Thinking…"):
|
||||
answer = query(cfg, q, max_tokens=500, verbose=verbose)
|
||||
console.print("[bold cyan]Standup:[/bold cyan]\n")
|
||||
console.print(answer)
|
||||
|
||||
|
||||
# ── status ────────────────────────────────────────────────────────
|
||||
|
||||
@cli.command()
|
||||
def status() -> None:
|
||||
"""Show config and state info."""
|
||||
try:
|
||||
cfg = load_config()
|
||||
except FileNotFoundError as e:
|
||||
console.print(f"[red]{e}[/red]")
|
||||
raise typer.Exit(1)
|
||||
|
||||
last = read_last_update(cfg)
|
||||
table = Table(show_header=False, box=None)
|
||||
table.add_row("[bold]config[/bold]", str(cfg.config_path))
|
||||
table.add_row("[bold]model[/bold]", str(cfg.model_path))
|
||||
table.add_row("[bold]state[/bold]",
|
||||
f"{cfg.state_path} ({cfg.state_path.stat().st_size:,} B)"
|
||||
if cfg.state_path.exists() else f"{cfg.state_path} (none)")
|
||||
table.add_row("[bold]scan paths[/bold]", ", ".join(str(p) for p in cfg.git.scan_paths))
|
||||
table.add_row("[bold]bash[/bold]", "on" if cfg.bash.enabled else "off")
|
||||
table.add_row("[bold]last update[/bold]",
|
||||
last.strftime("%Y-%m-%d %H:%M") if last else "never")
|
||||
if cfg.snapshots_dir.exists():
|
||||
snaps = sorted(cfg.snapshots_dir.glob("*.memb"))
|
||||
table.add_row("[bold]snapshots[/bold]",
|
||||
f"{len(snaps)} " + (f"(latest {snaps[-1].stem})" if snaps else ""))
|
||||
console.print(table)
|
||||
|
||||
|
||||
# ── prune ─────────────────────────────────────────────────────────
|
||||
|
||||
@cli.command()
|
||||
def prune(
|
||||
keep_days: int = typer.Option(90, "--keep-days", help="Snapshots older than this are deleted"),
|
||||
) -> None:
|
||||
"""Delete old daily snapshots."""
|
||||
cfg = load_config()
|
||||
n = prune_old(cfg, keep_days=keep_days)
|
||||
console.print(f"[dim]Deleted {n} snapshots older than {keep_days} days[/dim]")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
cli()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,146 @@
|
||||
"""Config & state-tracking for memwalk.
|
||||
|
||||
Layout:
|
||||
~/.memwalk/
|
||||
config.toml # user settings
|
||||
last_update.txt # ISO timestamp of last successful update
|
||||
current.memb # rolling state file
|
||||
snapshots/
|
||||
2026-05-16.memb
|
||||
...
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
if sys.version_info >= (3, 11):
|
||||
import tomllib
|
||||
else:
|
||||
import tomli as tomllib
|
||||
|
||||
|
||||
HOME_DIR = Path.home() / ".memwalk"
|
||||
|
||||
DEFAULT_MODEL_HINT = (
|
||||
"Recommended: NVIDIA Nemotron-3-Nano-4B (hybrid Mamba+Transformer).\n"
|
||||
"Download with: hf download nvidia/NVIDIA-Nemotron-3-Nano-4B-GGUF "
|
||||
"NVIDIA-Nemotron3-Nano-4B-Q4_K_M.gguf --local-dir ~/.memwalk/models"
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class GitConfig:
|
||||
scan_paths: list[Path] = field(default_factory=list)
|
||||
bootstrap_days: int = 30
|
||||
|
||||
|
||||
@dataclass
|
||||
class BashConfig:
|
||||
enabled: bool = True
|
||||
history_file: Path = Path.home() / ".bash_history"
|
||||
session_gap_min: int = 30
|
||||
|
||||
|
||||
@dataclass
|
||||
class Config:
|
||||
model_path: Path
|
||||
n_gpu_layers: int = -1
|
||||
n_ctx: int = 8192
|
||||
state_dir: Path = HOME_DIR
|
||||
git: GitConfig = field(default_factory=GitConfig)
|
||||
bash: BashConfig = field(default_factory=BashConfig)
|
||||
|
||||
@property
|
||||
def config_path(self) -> Path:
|
||||
return HOME_DIR / "config.toml"
|
||||
|
||||
@property
|
||||
def state_path(self) -> Path:
|
||||
return self.state_dir / "current.memb"
|
||||
|
||||
@property
|
||||
def snapshots_dir(self) -> Path:
|
||||
return self.state_dir / "snapshots"
|
||||
|
||||
@property
|
||||
def last_update_path(self) -> Path:
|
||||
return self.state_dir / "last_update.txt"
|
||||
|
||||
|
||||
def load_config() -> Config:
|
||||
"""Load config.toml from ~/.memwalk/ or raise FileNotFoundError."""
|
||||
path = HOME_DIR / "config.toml"
|
||||
if not path.exists():
|
||||
raise FileNotFoundError(
|
||||
f"No config at {path}. Run `memwalk init` first."
|
||||
)
|
||||
|
||||
with path.open("rb") as f:
|
||||
data = tomllib.load(f)
|
||||
|
||||
model_path = Path(data["model_path"]).expanduser()
|
||||
git_cfg = data.get("git", {})
|
||||
bash_cfg = data.get("bash", {})
|
||||
|
||||
return Config(
|
||||
model_path=model_path,
|
||||
n_gpu_layers=int(data.get("n_gpu_layers", -1)),
|
||||
n_ctx=int(data.get("n_ctx", 8192)),
|
||||
state_dir=Path(data.get("state_dir", HOME_DIR)).expanduser(),
|
||||
git=GitConfig(
|
||||
scan_paths=[Path(p).expanduser() for p in git_cfg.get("scan_paths", [])],
|
||||
bootstrap_days=int(git_cfg.get("bootstrap_days", 30)),
|
||||
),
|
||||
bash=BashConfig(
|
||||
enabled=bool(bash_cfg.get("enabled", True)),
|
||||
history_file=Path(bash_cfg.get("history_file",
|
||||
Path.home() / ".bash_history")).expanduser(),
|
||||
session_gap_min=int(bash_cfg.get("session_gap_min", 30)),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def write_config(cfg: Config) -> None:
|
||||
"""Persist config to ~/.memwalk/config.toml using a hand-rolled writer
|
||||
(Python stdlib has no TOML writer until 3.13+)."""
|
||||
HOME_DIR.mkdir(parents=True, exist_ok=True)
|
||||
cfg.state_dir.mkdir(parents=True, exist_ok=True)
|
||||
cfg.snapshots_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
lines = [
|
||||
"# memwalk config — edit by hand or rerun `memwalk init`",
|
||||
f'model_path = "{cfg.model_path}"',
|
||||
f"n_gpu_layers = {cfg.n_gpu_layers}",
|
||||
f"n_ctx = {cfg.n_ctx}",
|
||||
f'state_dir = "{cfg.state_dir}"',
|
||||
"",
|
||||
"[git]",
|
||||
"scan_paths = [",
|
||||
*[f' "{p}",' for p in cfg.git.scan_paths],
|
||||
"]",
|
||||
f"bootstrap_days = {cfg.git.bootstrap_days}",
|
||||
"",
|
||||
"[bash]",
|
||||
f"enabled = {str(cfg.bash.enabled).lower()}",
|
||||
f'history_file = "{cfg.bash.history_file}"',
|
||||
f"session_gap_min = {cfg.bash.session_gap_min}",
|
||||
]
|
||||
cfg.config_path.write_text("\n".join(lines) + "\n")
|
||||
|
||||
|
||||
def read_last_update(cfg: Config) -> datetime | None:
|
||||
path = cfg.last_update_path
|
||||
if not path.exists():
|
||||
return None
|
||||
try:
|
||||
return datetime.fromisoformat(path.read_text().strip())
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def write_last_update(cfg: Config, ts: datetime) -> None:
|
||||
cfg.last_update_path.write_text(ts.isoformat())
|
||||
@@ -0,0 +1,108 @@
|
||||
"""Orchestrate: pull events from sources, format, feed to a memba Session."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
from datetime import datetime, timedelta
|
||||
from pathlib import Path
|
||||
|
||||
from memba import Session
|
||||
|
||||
from .config import Config, read_last_update, write_last_update
|
||||
from .snapshot import maybe_snapshot
|
||||
from .sources import Event, bash as bash_src, git as git_src
|
||||
|
||||
|
||||
def open_session(cfg: Config, *, verbose: bool = False) -> Session:
|
||||
"""Open a memba Session pointing at memwalk's rolling state file."""
|
||||
return Session(
|
||||
model_path=str(cfg.model_path),
|
||||
session_id="current",
|
||||
state_dir=str(cfg.state_dir),
|
||||
n_gpu_layers=cfg.n_gpu_layers,
|
||||
n_ctx=cfg.n_ctx,
|
||||
chat_format="chatml",
|
||||
verbose=verbose,
|
||||
)
|
||||
|
||||
|
||||
def collect_all(cfg: Config, since: datetime) -> dict[str, list[Event]]:
|
||||
"""Pull events from every enabled source."""
|
||||
out: dict[str, list[Event]] = {}
|
||||
if cfg.git.scan_paths:
|
||||
out["git"] = git_src.collect(cfg.git.scan_paths, since)
|
||||
if cfg.bash.enabled:
|
||||
out["bash"] = bash_src.collect(
|
||||
cfg.bash.history_file, since, cfg.bash.session_gap_min
|
||||
)
|
||||
return out
|
||||
|
||||
|
||||
def format_block(events_by_source: dict[str, list[Event]],
|
||||
since: datetime, until: datetime) -> str:
|
||||
"""Build a single human-readable text block for ingestion."""
|
||||
chunks: list[str] = [
|
||||
f"=== Activity from {since.date()} to {until.date()} ==="
|
||||
]
|
||||
if events_by_source.get("git"):
|
||||
chunks.append(git_src.format_block(events_by_source["git"]))
|
||||
if events_by_source.get("bash"):
|
||||
chunks.append(bash_src.format_block(events_by_source["bash"]))
|
||||
return "\n\n".join(chunks)
|
||||
|
||||
|
||||
def update(cfg: Config, *, verbose: bool = False) -> dict:
|
||||
"""Ingest new events into the state and return a small summary dict."""
|
||||
now = datetime.now()
|
||||
since = read_last_update(cfg)
|
||||
if since is None:
|
||||
# First run: bootstrap window
|
||||
since = now - timedelta(days=cfg.git.bootstrap_days)
|
||||
|
||||
events = collect_all(cfg, since)
|
||||
n_git = len(events.get("git", []))
|
||||
n_bash = len(events.get("bash", []))
|
||||
|
||||
if n_git + n_bash == 0:
|
||||
return {"ingested": 0, "git": 0, "bash": 0, "since": since, "until": now}
|
||||
|
||||
block = format_block(events, since, now)
|
||||
|
||||
# Rotate daily snapshot BEFORE mutating state
|
||||
snapshotted = maybe_snapshot(cfg)
|
||||
|
||||
sess = open_session(cfg, verbose=verbose)
|
||||
|
||||
prompt = (
|
||||
"I will now feed you a summary of my recent work activity. "
|
||||
"Read it and incorporate into your understanding of what I've been "
|
||||
"doing. Reply with just 'noted'.\n\n" + block
|
||||
)
|
||||
|
||||
t0 = time.time()
|
||||
ack = sess.chat(prompt, max_tokens=20)
|
||||
elapsed = time.time() - t0
|
||||
|
||||
sess.save()
|
||||
write_last_update(cfg, now)
|
||||
|
||||
return {
|
||||
"ingested": n_git + n_bash,
|
||||
"git": n_git,
|
||||
"bash": n_bash,
|
||||
"since": since,
|
||||
"until": now,
|
||||
"ack": ack,
|
||||
"elapsed_s": elapsed,
|
||||
"snapshotted": snapshotted,
|
||||
"state_size": sess.state_size,
|
||||
}
|
||||
|
||||
|
||||
def query(cfg: Config, question: str, max_tokens: int = 400,
|
||||
verbose: bool = False) -> str:
|
||||
"""Load the current state and ask a question."""
|
||||
if not cfg.state_path.exists():
|
||||
raise FileNotFoundError("No state yet — run `memwalk update` first.")
|
||||
sess = open_session(cfg, verbose=verbose)
|
||||
return sess.chat(question, max_tokens=max_tokens)
|
||||
@@ -0,0 +1,36 @@
|
||||
"""Daily snapshot rotation for the rolling state file."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import shutil
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
from .config import Config
|
||||
|
||||
|
||||
def maybe_snapshot(cfg: Config) -> bool:
|
||||
"""If current.memb exists and no snapshot for today, copy it. Returns True if snapshotted."""
|
||||
if not cfg.state_path.exists():
|
||||
return False
|
||||
|
||||
today = datetime.now().strftime("%Y-%m-%d")
|
||||
target = cfg.snapshots_dir / f"{today}.memb"
|
||||
if target.exists():
|
||||
return False
|
||||
|
||||
cfg.snapshots_dir.mkdir(parents=True, exist_ok=True)
|
||||
shutil.copy2(cfg.state_path, target)
|
||||
return True
|
||||
|
||||
|
||||
def prune_old(cfg: Config, keep_days: int = 90) -> int:
|
||||
"""Delete snapshots older than @keep_days. Returns number deleted."""
|
||||
if not cfg.snapshots_dir.exists():
|
||||
return 0
|
||||
cutoff = (datetime.now() - timedelta(days=keep_days)).strftime("%Y-%m-%d")
|
||||
deleted = 0
|
||||
for snap in cfg.snapshots_dir.glob("*.memb"):
|
||||
if snap.stem < cutoff:
|
||||
snap.unlink()
|
||||
deleted += 1
|
||||
return deleted
|
||||
@@ -0,0 +1,13 @@
|
||||
"""Event sources — each module exposes `collect(since: datetime) -> list[Event]`."""
|
||||
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Event:
|
||||
"""A timestamped, human-readable activity record."""
|
||||
source: str # "git", "bash", "obsidian", ...
|
||||
ts: datetime
|
||||
summary: str # one-line description used in ingestion blocks
|
||||
detail: str = "" # optional longer body
|
||||
@@ -0,0 +1,153 @@
|
||||
"""
|
||||
Bash history source.
|
||||
|
||||
Parses ~/.bash_history, handling the optional HISTTIMEFORMAT prefix
|
||||
(`#<unix_ts>\\n<command>` blocks). Filters out noise (short / dupe / common
|
||||
navigation / secret-looking lines) and groups remaining commands into
|
||||
sessions separated by a configurable idle gap.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
from . import Event
|
||||
|
||||
# Commands too short or too common to be worth remembering
|
||||
_SKIP_EXACT = {
|
||||
"ls", "ll", "la", "l", "cd", "cd -", "pwd", "clear", "exit", "fg", "bg",
|
||||
"jobs", "history", "reset",
|
||||
}
|
||||
_SKIP_PREFIX = ("ls ", "cd ", "cat ", "less ", "tail ", "head ", "echo ",
|
||||
"which ", "whereis ", "type ", "man ", "help ")
|
||||
|
||||
# Lines that look like they leak credentials — never ingest
|
||||
_SECRET_PATTERNS = [
|
||||
re.compile(r"(?i)(password|passwd|secret|api[_-]?key|access[_-]?token|bearer)\s*[=:]"),
|
||||
re.compile(r"(?i)\b(aws|gcp|gh|github|hf|huggingface|openai|anthropic)[_-]?(token|key)"),
|
||||
re.compile(r"(?i)sk-[a-z0-9-]{20,}"), # OpenAI/Anthropic API keys
|
||||
re.compile(r"(?i)ghp_[a-z0-9]{30,}"), # GitHub PATs
|
||||
]
|
||||
|
||||
|
||||
def _is_noise(cmd: str) -> bool:
|
||||
if len(cmd) < 3:
|
||||
return True
|
||||
if cmd in _SKIP_EXACT:
|
||||
return True
|
||||
for p in _SKIP_PREFIX:
|
||||
if cmd.startswith(p) and len(cmd) < 25:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _looks_secret(cmd: str) -> bool:
|
||||
return any(p.search(cmd) for p in _SECRET_PATTERNS)
|
||||
|
||||
|
||||
@dataclass
|
||||
class _Raw:
|
||||
ts: datetime | None
|
||||
cmd: str
|
||||
|
||||
|
||||
def _parse_history(path: Path) -> list[_Raw]:
|
||||
"""Read history file, returning entries with timestamps where available."""
|
||||
if not path.exists():
|
||||
return []
|
||||
|
||||
try:
|
||||
lines = path.read_text(encoding="utf-8", errors="replace").splitlines()
|
||||
except OSError:
|
||||
return []
|
||||
|
||||
entries: list[_Raw] = []
|
||||
pending_ts: datetime | None = None
|
||||
for line in lines:
|
||||
if not line:
|
||||
continue
|
||||
if line.startswith("#") and line[1:].strip().isdigit():
|
||||
# HISTTIMEFORMAT marker
|
||||
try:
|
||||
pending_ts = datetime.fromtimestamp(int(line[1:].strip()))
|
||||
except (ValueError, OSError):
|
||||
pending_ts = None
|
||||
continue
|
||||
entries.append(_Raw(ts=pending_ts, cmd=line.strip()))
|
||||
pending_ts = None
|
||||
return entries
|
||||
|
||||
|
||||
def collect(
|
||||
history_file: Path,
|
||||
since: datetime,
|
||||
session_gap_min: int = 30,
|
||||
) -> list[Event]:
|
||||
"""Return cleaned bash events newer than @since, grouped into sessions."""
|
||||
raw = _parse_history(history_file)
|
||||
if not raw:
|
||||
return []
|
||||
|
||||
# Filter
|
||||
clean: list[_Raw] = []
|
||||
prev_cmd: str | None = None
|
||||
for entry in raw:
|
||||
cmd = entry.cmd
|
||||
if not cmd or cmd.startswith("#"):
|
||||
continue
|
||||
if _is_noise(cmd) or _looks_secret(cmd):
|
||||
continue
|
||||
if cmd == prev_cmd:
|
||||
continue
|
||||
if entry.ts is not None and entry.ts < since:
|
||||
continue
|
||||
clean.append(entry)
|
||||
prev_cmd = cmd
|
||||
|
||||
if not clean:
|
||||
return []
|
||||
|
||||
# Group by session (gap > session_gap_min minutes starts a new one).
|
||||
# Commands without timestamps are attributed to the previous session.
|
||||
events: list[Event] = []
|
||||
session: list[_Raw] = []
|
||||
last_ts: datetime | None = None
|
||||
fallback_ts = since # used when entries have no timestamps
|
||||
|
||||
def flush_session(items: list[_Raw]) -> None:
|
||||
if not items:
|
||||
return
|
||||
ts = next((it.ts for it in items if it.ts is not None), fallback_ts)
|
||||
commands = [it.cmd for it in items]
|
||||
summary = f"shell session ({len(commands)} cmds): {commands[0][:60]}"
|
||||
detail = "\n".join(f" $ {c}" for c in commands[:30])
|
||||
if len(commands) > 30:
|
||||
detail += f"\n … and {len(commands)-30} more"
|
||||
events.append(Event(source="bash", ts=ts, summary=summary, detail=detail))
|
||||
|
||||
gap = session_gap_min * 60
|
||||
for entry in clean:
|
||||
if entry.ts is not None and last_ts is not None:
|
||||
if (entry.ts - last_ts).total_seconds() > gap:
|
||||
flush_session(session)
|
||||
session = []
|
||||
session.append(entry)
|
||||
if entry.ts is not None:
|
||||
last_ts = entry.ts
|
||||
flush_session(session)
|
||||
return events
|
||||
|
||||
|
||||
def format_block(events: list[Event]) -> str:
|
||||
if not events:
|
||||
return ""
|
||||
parts = ["shell activity:"]
|
||||
for e in events:
|
||||
date = e.ts.strftime("%Y-%m-%d %H:%M")
|
||||
parts.append(f"\n [{date}] {e.summary}")
|
||||
if e.detail:
|
||||
parts.append(e.detail)
|
||||
return "\n".join(parts)
|
||||
@@ -0,0 +1,92 @@
|
||||
"""Git event source — walks configured paths for repos and collects commits."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
from . import Event
|
||||
|
||||
|
||||
def find_repos(roots: list[Path]) -> list[Path]:
|
||||
"""Return all distinct git repositories under the given roots (non-recursive
|
||||
one level deep, plus the root itself if it's a repo)."""
|
||||
repos: list[Path] = []
|
||||
seen: set[Path] = set()
|
||||
for root in roots:
|
||||
if not root.exists():
|
||||
continue
|
||||
candidates: list[Path] = []
|
||||
if (root / ".git").exists():
|
||||
candidates.append(root)
|
||||
for entry in root.iterdir():
|
||||
if entry.is_dir() and (entry / ".git").exists():
|
||||
candidates.append(entry)
|
||||
for c in candidates:
|
||||
real = c.resolve()
|
||||
if real not in seen:
|
||||
seen.add(real)
|
||||
repos.append(c)
|
||||
return repos
|
||||
|
||||
|
||||
def collect(roots: list[Path], since: datetime) -> list[Event]:
|
||||
"""Collect commits across @roots that landed after @since."""
|
||||
since_iso = since.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S")
|
||||
events: list[Event] = []
|
||||
|
||||
for repo in find_repos(roots):
|
||||
try:
|
||||
out = subprocess.check_output(
|
||||
[
|
||||
"git", "-C", str(repo), "log",
|
||||
f"--since={since_iso}",
|
||||
"--no-merges",
|
||||
"--date=iso-strict",
|
||||
"--pretty=format:%H%x09%ad%x09%an%x09%s",
|
||||
],
|
||||
text=True, stderr=subprocess.DEVNULL,
|
||||
)
|
||||
except subprocess.CalledProcessError:
|
||||
continue
|
||||
|
||||
for line in out.splitlines():
|
||||
parts = line.split("\t", 3)
|
||||
if len(parts) != 4:
|
||||
continue
|
||||
sha, iso, author, subject = parts
|
||||
try:
|
||||
ts = datetime.fromisoformat(iso)
|
||||
except ValueError:
|
||||
continue
|
||||
events.append(Event(
|
||||
source="git",
|
||||
ts=ts,
|
||||
summary=f"[{repo.name}] {subject}",
|
||||
detail=f"commit {sha[:10]} by {author}",
|
||||
))
|
||||
|
||||
return events
|
||||
|
||||
|
||||
def format_block(events: list[Event]) -> str:
|
||||
"""Group commits by repo (extracted from `[name] subject`) into a readable block."""
|
||||
if not events:
|
||||
return ""
|
||||
|
||||
by_repo: dict[str, list[Event]] = {}
|
||||
for e in events:
|
||||
# Strip leading "[repo] " prefix to recover repo name
|
||||
repo = e.summary.split("]", 1)[0].lstrip("[") if e.summary.startswith("[") else "unknown"
|
||||
by_repo.setdefault(repo, []).append(e)
|
||||
|
||||
parts: list[str] = ["git activity:"]
|
||||
for repo, items in sorted(by_repo.items()):
|
||||
items.sort(key=lambda x: x.ts)
|
||||
parts.append(f"\n {repo} ({len(items)} commits):")
|
||||
for e in items:
|
||||
date = e.ts.strftime("%Y-%m-%d")
|
||||
subject = e.summary.split("] ", 1)[-1]
|
||||
parts.append(f" {date} {subject}")
|
||||
return "\n".join(parts)
|
||||
@@ -0,0 +1,52 @@
|
||||
[build-system]
|
||||
requires = ["setuptools>=68", "wheel"]
|
||||
build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "memwalk"
|
||||
version = "0.1.0"
|
||||
description = "Walk through your work memory — semantic recall of git, shell and notes via local SSM models"
|
||||
readme = "README.md"
|
||||
license = { text = "MIT" }
|
||||
requires-python = ">=3.10"
|
||||
|
||||
keywords = ["llm", "memory", "ssm", "mamba", "nemotron", "memba", "personal-ai", "cli"]
|
||||
classifiers = [
|
||||
"Development Status :: 3 - Alpha",
|
||||
"Environment :: Console",
|
||||
"Intended Audience :: Developers",
|
||||
"License :: OSI Approved :: MIT License",
|
||||
"Programming Language :: Python :: 3",
|
||||
"Programming Language :: Python :: 3.10",
|
||||
"Programming Language :: Python :: 3.11",
|
||||
"Programming Language :: Python :: 3.12",
|
||||
"Topic :: Utilities",
|
||||
]
|
||||
|
||||
dependencies = [
|
||||
"memba>=0.1.0",
|
||||
"typer>=0.9.0",
|
||||
"rich>=13.0.0",
|
||||
"tomli>=2.0;python_version<'3.11'",
|
||||
]
|
||||
|
||||
[project.optional-dependencies]
|
||||
dev = ["pytest>=7.0", "ruff", "mypy"]
|
||||
|
||||
[project.scripts]
|
||||
memwalk = "memwalk.cli:main"
|
||||
|
||||
[project.urls]
|
||||
Homepage = "https://github.com/emil28092005/memwalk"
|
||||
Repository = "https://github.com/emil28092005/memwalk"
|
||||
|
||||
[tool.setuptools.packages.find]
|
||||
include = ["memwalk*"]
|
||||
|
||||
[tool.ruff]
|
||||
line-length = 100
|
||||
target-version = "py310"
|
||||
|
||||
[tool.ruff.lint]
|
||||
select = ["E", "F", "W", "I"]
|
||||
ignore = ["E501"]
|
||||
Reference in New Issue
Block a user