cli

The kosha shell command: every index and graph query as a subcommand.

This module provides one kosha command, for agents that work through a shell and for people at a terminal. Each query has a subcommand: sync, status, context, repo-context, env-context, ni, public-api, api-paths, dep-stack, top-nodes, where-to-add, watch, daemon, install, nuke, embed and prune. Subcommand names and flags use hyphens where the Python methods use underscores, as in --as-json and --busy-timeout. A flag that is on by default turns off with --no- in front, as in kosha sync --no-graph.

Output is readable markdown. --as-json prints JSON instead, for piping into jq or another program. kosha <subcommand> --help lists a subcommand’s flags.

from fastcore.foundation import L
assert _to_json({'a': {1, 2}}) == {'a': [1, 2]}
assert _to_json(L([{'x': 1}])) == [{'x': 1}]
assert _to_json('plain') == 'plain'
print("helpers ok")
helpers ok

kosha sync brings the index up to date, with the same switches as Kosha.sync: --no-repo, --no-env and --no-graph skip a part, --pkgs names the packages to index as a comma-separated list, and --force re-reads everything. To index packages in parallel, pass both --pkg-parallel and --busy-timeout 30000. When it finishes it prints the index status.

sync

def sync(
    dir:str=None, # directory to sync; defaults to repo root
    pkgs:str=None, # comma-separated package names; defaults to all pyproject.toml deps
    repo:bool=True, # index the repo
    env:bool=True, # index installed packages; --no-env indexes the repo only
    graph:bool=True, # build the call graph
    graph_mode:str='fast', # 'full' re-extracts changed files in worker processes
    graph_metrics:bool=True, # recompute PageRank after the graph update
    parallel:bool=True, # run repo, env, and graph sync in parallel
    embed:bool=True, # compute embeddings; --no-embed stores chunks for keyword search only
    force:bool=False, # re-sync everything, graph included, rather than what changed
    busy_timeout:int=None, # ms to wait on a locked db; set (e.g. 30000) when syncing in parallel
    pkg_parallel:bool=False, # index packages in parallel threads; needs --busy-timeout
    as_json:bool=False, # output the post-sync status as JSON
):

Index the repo, the installed packages and the call graph, then print the index status.

from fastcore.test import test_eq

test_eq(_split_pkgs('fastcore,httpx'), ['fastcore', 'httpx'])
# commas inside extras belong to the extras, not to the list
test_eq(_split_pkgs('rishi[litert,llama,all]'), ['rishi'])
test_eq(_split_pkgs('rishi[litert,llama,all],fastcore'), ['rishi', 'fastcore'])
test_eq(_split_pkgs('rishi[all]>=0.2, fastcore'), ['rishi>=0.2', ' fastcore'])
# empty segments would crash `_pkg_name`, so they never make it out
test_eq(_split_pkgs('fastcore,,'), ['fastcore'])
print('_split_pkgs tests passed')
_split_pkgs tests passed

kosha context "query" runs Kosha.context, the combined search over the repo and the packages, and prints each hit with its name, line, PageRank, the first 300 characters of its source and up to five callers. The query can carry key:value filters. A filter-shaped token with an unknown key, such as libr:litesearch, prints a warning on standard error. kosha searches that token as text and does not filter with it. repo-context and env-context search one index only.

context

def context(
    query:str, # search query (supports key:value filters e.g. package:fastcore)
    limit:int=10, # max results
    repo:bool=True, # include repo results
    env:bool=True, # include env results
    graph:bool=True, # include call graph enrichment
    rerank:bool=False, # reorder the top-k with a flashrank cross-encoder
    as_json:bool=False, # output JSON instead of markdown
):

Search the repo and installed packages together and print the hits with their call-graph details.

assert callable(sync)
assert callable(context)
print("sync + context defined ok")

# `sync` and `context` use the same flag names `repo`, `env` and `graph`.
# `sync` has no `--sync_graph` flag. The call graph is on by default, so `ni` and `where-to-add` have a graph to read.
from fastcore.script import anno_parser
_flags = {a.dest for a in anno_parser(sync, prog='kosha sync')._actions}
assert {'repo', 'env', 'graph', 'graph_mode', 'graph_metrics'} <= _flags, sorted(_flags)
assert 'sync_graph' not in _flags, sorted(_flags)
_ctx = {a.dest for a in anno_parser(context, prog='kosha context')._actions}
assert {'repo', 'env', 'graph'} <= _ctx, sorted(_ctx)

# `sync` passes `graph_mode` and `graph_metrics` to `Kosha.sync` unchanged.
_cli_calls = []
class _CliKosha:
    def __init__(self, **kwargs): pass
    def sync(self, **kwargs): _cli_calls.append(kwargs)
    def status(self): return {}
_cli_kosha, sync.__wrapped__.__globals__['Kosha'] = sync.__wrapped__.__globals__['Kosha'], _CliKosha
try: sync(graph_mode='full', graph_metrics=False)
finally: sync.__wrapped__.__globals__['Kosha'] = _cli_kosha
test_eq((_cli_calls[0]['graph_mode'], _cli_calls[0]['graph_metrics']), ('full', False))
sync + context defined ok

repo_context

def repo_context(
    query:str, # search query
    limit:int=10, # max results
    as_json:bool=False, # output JSON
):

Search the repo index only and print the hits.

env_context

def env_context(
    query:str, # search query (supports key:value filters e.g. package:fastcore)
    limit:int=10, # max results
    as_json:bool=False, # output JSON
):

Search the installed-package index only and print the hits.

The graph subcommands take full dotted names. kosha ni fastcore.basics.merge prints a node’s PageRank and up to ten of its callers, callees and co-dispatched peers. public-api, api-paths, dep-stack and top-nodes print the results of the Kosha methods of those names; dep-stack with no --seeds starts from the packages in pyproject.toml. kosha watch re-indexes files as they change, until Ctrl-C.

ni

def ni(
    mod_name:str, # full dotted node name, e.g. fastcore.basics.merge
    as_json:bool=False, # output JSON
):

Print the PageRank, callers, callees and co-dispatched peers of one graph node.

watch

def watch(
    dir:str=None, # directory to watch; defaults to repo root
):

Watch the repo and re-index files as they change, until Ctrl-C.

public_api

def public_api(
    pkg:str, # package name e.g. "fastcore" or submodule "fastcore.basics"
    module:str=None, # restrict to a submodule (overrides pkg for scoping)
    as_json:bool=False, # output JSON
):

Print the public names of a package or module, including functions @patch adds to its classes.

api_paths

def api_paths(
    from_pkg:str, # source package (call origin)
    to_pkg:str, # target package (call destination)
    k:int=15, # top-k API nodes per package to consider
    as_json:bool=False, # output JSON
):

Print the shortest call paths from the top public functions of from_pkg to those of to_pkg.

dep_stack

def dep_stack(
    seeds:str=None, # comma-separated seed package names; defaults to pyproject.toml deps
    depth:int=1, # number of layers below the seeds
    as_json:bool=False, # output JSON
):

Print the layers of packages imported by the seed packages, breadth first.

top_nodes

def top_nodes(
    pkg:str, # package name e.g. "fastcore"
    k:int=5, # number of top nodes to return
    as_json:bool=False, # output JSON
):

Print the k public names of a package with the highest PageRank.

status

def status(
    as_json:bool=False
):

Print how fresh the index is: file, package and node counts, stale files and stale packages.

kosha where-to-add "description" prints path:line suggestions for where new code should go, with the peers it would join. kosha nuke deletes the repo’s index and call graph, and kosha nuke --env also deletes the shared package and docs indexes.

where_to_add

def where_to_add(
    description:str, # what you want to add
    limit:int=5, # max results
    as_json:bool=False, # output JSON
):

Print the files and lines where code matching description would best be added.

nuke

def nuke(
    env:bool=False, # also delete the shared package and docs indexes
):

Delete the repo index and call graph, and with –env the shared indexes too.

kosha embed embeds the package chunks that sync --no-embed or an older kosha stored without an embedding. It works one batch at a time. Stopping it loses at most one batch. kosha prune removes the packages and versions that no indexed environment still has installed and compacts the index; --dry-run lists them without removing anything.

prune

def prune(
    dry_run:bool=False, # list what would be removed without removing it
    as_json:bool=False, # output JSON
):

Remove the packages and versions no indexed environment still has installed, and print what was removed.

embed

def embed(
    pkg:str=None, # only this package; defaults to every package
    batch:int=5000, # chunks embedded, committed and indexed together
):

Embed the package chunks that have no embedding yet and print how many were embedded.

kosha daemon keeps one Kosha loaded. Repeated queries then skip the few seconds that loading the embedding model takes. It prints {"ok": true, "status": "ready"}. It then reads one JSON request per line from standard input, such as {"cmd": "context", "args": {"q": "embed a query"}}, and writes one JSON line per answer: {"ok": true, "result": ...} or {"ok": false, "error": "..."}. The commands are the keys of _DISPATCH below, and their arguments are those of the Kosha methods of the same names. kosha install copies SKILL.md into the repository for coding agents.

daemon

def daemon():

Answer JSON requests from stdin, one per line, with one JSON line each on stdout, keeping kosha loaded.

install

def install():

Copy the bundled SKILL.md into .agents/skills/kosha/ and .claude/skills/kosha/ in the repo.

main is the entry point behind the kosha command. It looks up the subcommand in CMDS, parses the remaining arguments against that function’s signature, and calls it. With no subcommand, or an unknown one, it prints the list of subcommands.

main

def main():

Entry point for the kosha CLI command.

assert set(CMDS.keys()) == {
    'sync', 'context', 'repo-context', 'env-context', 'ni',
    'watch', 'public-api', 'api-paths', 'dep-stack', 'top-nodes',
    'daemon', 'status', 'where-to-add', 'install', 'nuke', 'embed', 'prune'
}
print(f"CMDS registered: {sorted(CMDS)}")
CMDS registered: ['api-paths', 'context', 'daemon', 'dep-stack', 'env-context', 'install', 'ni', 'public-api', 'repo-context', 'status', 'sync', 'top-nodes', 'watch', 'where-to-add']