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
    parallel:bool=False, # run repo, env, and graph sync in parallel
    embed:bool=True, # embed code chunks (set False for fast metadata-only update)
    force:bool=False, # force re-sync of all files (ignoring freshness)
    force_graph:bool=False, # force re-sync of call graph (ignoring freshness
):

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

---

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

---

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.

---

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'
}
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']