cli

Single kosha entry point with subcommands for shell-based harnesses.

from fastcore.foundation import L

# Test _to_json handles all expected types
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

---

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 env packages; --no-env is "this repo only", and skips the staleness scan
    graph:bool=True, # build the call graph
    parallel:bool=True, # run repo, env, and graph sync in parallel
    embed:bool=True, # embed code chunks (set False for fast metadata-only update)
    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, # ingest env packages concurrently; requires --busy_timeout
    as_json:bool=False, # output the post-sync status as JSON
):

Sync repo + env packages + call graph into .kosha/ databases.

The flags are context’s flags, and they mean the same thing here. They used to be one --sync_graph that defaulted to off, so this command promised a call graph in its own help text and never built one — kosha ni then found nothing and there was no way to tell why.

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

---

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
):

Fan-out semantic search over repo and env, optionally graph-enriched.

# Test that functions are callable
assert callable(sync)
assert callable(context)
print("sync + context defined ok")

# `sync` and `context` name the same three legs, so one command's flags read across to the other.
# `sync` used to take a single `--sync_graph` that was off by default, which made its own help text
# ("repo + env packages + call graph") false and left `ni`/`where-to-add` with no 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'} <= _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 + context defined ok

---

repo_context

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

Semantic + keyword search over indexed repo code only.

---

env_context

def env_context(
    query:str, # search query (package names auto-detected as filters)
    limit:int=10, # max results
    as_json:bool=False, # output JSON
):

Semantic search over indexed env packages only.

---

ni

def ni(
    mod_name:str, # fully-qualified module node name e.g. fastcore.basics.merge
    as_json:bool=False, # output JSON
):

Node info: callers, callees, co_dispatched, pagerank for a single graph node.

---

watch

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

Live incremental re-index on file changes. Blocking — Ctrl-C to stop.

---

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
):

List public API entries for a package (respects all + @patch methods).

---

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
):

Shortest call-graph paths from from_pkg public API to to_pkg public API.

---

dep_stack

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

BFS dependency layers from seed packages, ordered by coupling strength.

---

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
):

Top-k public API nodes for a package ranked by PageRank in the call graph.

---

status

def status(
    as_json:bool=False
):

Show index freshness: file/pkg/node counts, stale files, and stale packages.

---

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
):

Find likely insertion points for new code matching description.

---

nuke

def nuke(
    env:bool=False, # delete env cache
):

Delete all kosha data and caches.

---

daemon

def daemon():

Persistent kosha kernel. Reads newline-delimited JSON from stdin, writes results to stdout.

---

install

def install():

Install kosha SKILL.md to .agents/skills/kosha/ and .claude/skills/kosha/.

---

main

def main():

Entry point for the kosha CLI command.

# Test dispatcher has all expected commands
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'
}
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']