memory_cmd

memory_cmd

jarvis memory — memory management subcommands.

Classes

Functions

memory

memory() -> None

Manage the memory store.

Source code in src/openjarvis/cli/memory_cmd.py

@click.group()
def memory() -> None:
    """Manage the memory store."""

index

index(path: str, backend: str | None, chunk_size: int, chunk_overlap: int) -> None

Index documents from a file or directory.

Source code in src/openjarvis/cli/memory_cmd.py

@memory.command()
@click.argument("path")
@click.option(
    "--backend",
    "-b",
    default=None,
    help="Override the default memory backend.",
)
@click.option(
    "--chunk-size",
    default=512,
    type=int,
    help="Chunk size in tokens.",
)
@click.option(
    "--chunk-overlap",
    default=64,
    type=int,
    help="Overlap between chunks in tokens.",
)
def index(
    path: str,
    backend: str | None,
    chunk_size: int,
    chunk_overlap: int,
) -> None:
    """Index documents from a file or directory."""
    console = Console(stderr=True)
    target = Path(path)

    if not target.exists():
        console.print(f"[red]Path not found:[/red] {path}")
        raise SystemExit(1)

    t0 = time.time()
    cfg = ChunkConfig(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
    )

    console.print(f"[cyan]Indexing[/cyan] {path} ...")
    chunks = ingest_path(target, config=cfg)

    if not chunks:
        console.print("[yellow]No indexable content found.[/yellow]")
        return

    mem = _get_backend(backend)
    try:
        replace_source = getattr(mem, "replace_source", None)
        if callable(replace_source):
            documents_by_source = {}
            for chunk in chunks:
                documents_by_source.setdefault(chunk.source, []).append(
                    (
                        chunk.content,
                        {
                            "offset": chunk.offset,
                            "index": chunk.index,
                        },
                    )
                )
            for source, documents in track(
                documents_by_source.items(),
                description="Replacing sources...",
                console=console,
            ):
                replace_source(source, documents)
        else:
            for chunk in track(
                chunks,
                description="Storing chunks...",
                console=console,
            ):
                mem.store(
                    chunk.content,
                    source=chunk.source,
                    metadata={
                        "offset": chunk.offset,
                        "index": chunk.index,
                    },
                )
    finally:
        if hasattr(mem, "close"):
            mem.close()

    elapsed = time.time() - t0
    sources = {c.source for c in chunks}
    console.print(
        f"[green]Indexed {len(chunks)} chunks "
        f"from {len(sources)} file(s) "
        f"in {elapsed:.1f}s.[/green]"
    )

search

search(query: tuple[str, ...], top_k: int, backend: str | None) -> None

Search the memory store.

Source code in src/openjarvis/cli/memory_cmd.py

@memory.command()
@click.argument("query", nargs=-1, required=True)
@click.option(
    "--top-k",
    "-k",
    default=5,
    type=int,
    help="Number of results to return.",
)
@click.option(
    "--backend",
    "-b",
    default=None,
    help="Override the default memory backend.",
)
def search(
    query: tuple[str, ...],
    top_k: int,
    backend: str | None,
) -> None:
    """Search the memory store."""
    console = Console()
    query_text = " ".join(query)

    mem = _get_backend(backend)
    try:
        results = mem.retrieve(query_text, top_k=top_k)
    finally:
        if hasattr(mem, "close"):
            mem.close()

    if not results:
        console.print("[yellow]No results found.[/yellow]")
        return

    table = Table(title=f"Search: {query_text}")
    table.add_column("#", style="dim", width=3)
    table.add_column("Score", width=8)
    table.add_column("Source", style="cyan")
    table.add_column("Content")

    for i, r in enumerate(results, 1):
        # Truncate content for display
        preview = r.content[:200]
        if len(r.content) > 200:
            preview += "..."
        table.add_row(
            str(i),
            f"{r.score:.4f}",
            r.source or "-",
            preview,
        )

    console.print(table)

list_facts

list_facts() -> None

List durable facts captured by the automatic memory service.

Source code in src/openjarvis/cli/memory_cmd.py

@memory.command(name="list")
def list_facts() -> None:
    """List durable facts captured by the automatic memory service."""
    console = Console()

    store = _get_fact_store()
    facts = store.list()
    if not facts:
        console.print("[yellow]No memory facts stored yet.[/yellow]")
        return

    table = Table(title=f"Memory Facts ({len(facts)})")
    table.add_column("#", style="dim", width=4)
    table.add_column("Fact")
    table.add_column("Source", style="cyan")
    table.add_column("Trust")
    any_quarantined = False
    for i, fact in enumerate(facts, 1):
        quarantined = not fact.trusted_for_recall
        any_quarantined = any_quarantined or quarantined
        trust_disp = (
            "[red]⚠ quarantined[/red]"
            if quarantined
            else f"[green]{fact.trust or 'legacy'}[/green]"
        )
        table.add_row(str(i), fact.text, fact.source or "-", trust_disp)
    console.print(table)
    if any_quarantined:
        console.print(
            "[red]⚠ quarantined[/red] facts are retained for audit but excluded "
            "from model recall because they may contain hostile input. Review "
            "one and run [bold]jarvis memory trust <#>[/bold] to allow recall."
        )

trust_fact

trust_fact(index: int) -> None

Promote quarantined fact INDEX (as shown by memory list) to trusted.

Only do this after reading the fact: a quarantined fact matched an injection pattern, and promoting it puts its text into future prompts.

Source code in src/openjarvis/cli/memory_cmd.py

@memory.command(name="trust")
@click.argument("index", type=int)
def trust_fact(index: int) -> None:
    """Promote quarantined fact INDEX (as shown by `memory list`) to trusted.

    Only do this after reading the fact: a quarantined fact matched an
    injection pattern, and promoting it puts its text into future prompts.
    """
    console = Console()

    store = _get_fact_store()
    facts = store.list()
    if not 1 <= index <= len(facts):
        console.print(
            f"[red]No fact #{index}.[/red] "
            f"{len(facts)} fact(s) stored — see [bold]jarvis memory list[/bold]."
        )
        raise SystemExit(1)

    fact = facts[index - 1]
    if fact.trusted_for_recall:
        console.print(f"[yellow]Fact #{index} is already recallable.[/yellow]")
        return
    if not store.promote_reviewed(index - 1, fact.text):
        console.print(f"[red]Could not update fact #{index}.[/red]")
        raise SystemExit(1)
    console.print(f"[green]Fact #{index} is now trusted for recall:[/green]")
    console.print(f"  {fact.text}")

clear

clear(yes: bool) -> None

Remove all durable facts captured by the automatic memory service.

Source code in src/openjarvis/cli/memory_cmd.py

@memory.command()
@click.option(
    "--yes",
    "-y",
    is_flag=True,
    default=False,
    help="Skip the confirmation prompt.",
)
def clear(yes: bool) -> None:
    """Remove all durable facts captured by the automatic memory service."""
    console = Console()

    store = _get_fact_store()
    count = store.count()
    if count == 0:
        console.print("[yellow]No memory facts to clear.[/yellow]")
        return

    if not yes:
        if not click.confirm(f"Remove all {count} stored memory fact(s)?"):
            console.print("[dim]Aborted.[/dim]")
            return

    removed = store.clear()
    console.print(f"[green]Cleared {removed} memory fact(s).[/green]")

stats

stats(backend: str | None) -> None

Show memory store statistics.

Source code in src/openjarvis/cli/memory_cmd.py

@memory.command()
@click.option(
    "--backend",
    "-b",
    default=None,
    help="Override the default memory backend.",
)
def stats(backend: str | None) -> None:
    """Show memory store statistics."""
    console = Console()

    mem = _get_backend(backend)
    try:
        count = 0
        if hasattr(mem, "count"):
            count = mem.count()

        table = Table(title="Memory Statistics")
        table.add_column("Property", style="cyan")
        table.add_column("Value")
        table.add_row("Backend", mem.backend_id)
        table.add_row("Documents", str(count))

        if hasattr(mem, "_db_path"):
            db_path = Path(mem._db_path)
            if db_path.exists():
                size_kb = db_path.stat().st_size / 1024
                table.add_row(
                    "Database size",
                    f"{size_kb:.1f} KB",
                )
            table.add_row("Database path", str(db_path))

        console.print(table)
    finally:
        if hasattr(mem, "close"):
            mem.close()