API Reference

juplit

juplit.tasks

Core notebook workflow tasks for juplit.

These functions back both the poe task targets and the juplit CLI commands. They can also be imported and called directly from Python.

sync_notebooks()

Sync .py and .ipynb files for all paired percent-format notebooks.

Walks the configured notebook_src_dirs (from [tool.juplit] in pyproject.toml) and calls jupytext --sync on every .py file that has a jupytext percent-format header pairing it with an .ipynb.

Prints a summary that splits synced files by direction (py → ipynb vs ipynb → py), alongside unchanged and skipped files. Also flags any conflict — a file whose .py and .ipynb both changed since the last sync, where jupytext keeps the newer by mtime and overwrites the other.

Protects a .py that has edits since the last sync from being silently overwritten by a newer (mtime) .ipynb: such files are run with jupytext's --check-source-is-newer guard, so instead of clobbering the .py they are left untouched and reported as overwrite blocked. Files whose .py is unchanged since the last sync still sync both ways as usual.

Raises SystemExit(1) if jupytext reports any errors, or if any file was overwrite-blocked.

Source code in juplit/tasks.py

def sync_notebooks() -> None:
    """Sync `.py` and `.ipynb` files for all paired percent-format notebooks.

    Walks the configured `notebook_src_dirs` (from `[tool.juplit]` in
    `pyproject.toml`) and calls `jupytext --sync` on every `.py` file that has
    a jupytext percent-format header pairing it with an `.ipynb`.

    Prints a summary that splits synced files by direction (`py → ipynb` vs
    `ipynb → py`), alongside unchanged and skipped files. Also flags any
    *conflict* — a file whose `.py` and `.ipynb` both changed since the last
    sync, where jupytext keeps the newer by mtime and overwrites the other.

    Protects a `.py` that has edits since the last sync from being silently
    overwritten by a newer (mtime) `.ipynb`: such files are run with jupytext's
    `--check-source-is-newer` guard, so instead of clobbering the `.py` they are
    left untouched and reported as *overwrite blocked*. Files whose `.py` is
    unchanged since the last sync still sync both ways as usual.

    Raises `SystemExit(1)` if jupytext reports any errors, or if any file was
    overwrite-blocked.
    """
    from juplit.artifacts import artifact_py_files

    files = sorted(set(_find_percent_notebook_py_files()) | set(artifact_py_files()))
    if not files:
        print("No percent notebook .py files found.")
        return

    _prepare_artifacts(files)
    prev = _load_hashes()
    root = _repo_root()
    guarded = [f for f in files if _py_edited_since_sync(f, root, prev)]
    normal = [f for f in files if f not in guarded]

    g_normal, e_normal = _run_jupytext(["--sync"], normal)
    g_guard, e_guard = _run_jupytext(
        ["--sync", "--check-source-is-newer", "--warn-only"], guarded
    )
    groups = {k: g_normal[k] + g_guard[k] for k in g_normal}
    errors = e_normal + e_guard

    # Blocked files were not synced — keep their prior baseline so the guard
    # keeps firing until the user resolves them; save everything else once.
    risk_keys = set(groups["overwrite_risk"])
    _normalize_artifacts([f for f in files if _key(f, root) not in risk_keys])
    _save_hashes([f for f in files if _key(f, root) not in risk_keys])
    _warn_gitignored_artifacts(files)

    if groups["to_ipynb"]:
        print(_fmt("synced py → ipynb", groups["to_ipynb"]))
    if groups["to_py"]:
        print(_fmt("synced ipynb → py", groups["to_py"]))
    if groups["unchanged"]:
        print(_fmt("sync unchanged", groups["unchanged"]))
    if groups["skipped"]:
        print(_fmt("sync skipped (not paired)", groups["skipped"]))
    if groups["conflicts"]:
        print(_fmt(
            "sync CONFLICT — both .py and .ipynb changed since last sync; "
            "jupytext kept the newer by mtime and overwrote the other",
            groups["conflicts"],
        ))
    if groups["overwrite_risk"]:
        print(_fmt(
            "sync OVERWRITE BLOCKED — .py has unsynced edits but its .ipynb is "
            "newer; refused to overwrite the .py (resolve, e.g. delete the stale "
            ".ipynb or touch the .py, then re-sync)",
            groups["overwrite_risk"],
        ))
    if not any(groups.values()):
        print("Sync: nothing to do")
    for err in errors:
        print(f"sync error: {err}")
    stale = _report_stale_artifacts(files)
    if errors or groups["overwrite_risk"] or stale:
        raise SystemExit(1)

generate_notebooks()

Generate .ipynb files from .py percent-format files.

Calls jupytext --to notebook on every paired .py file found in the configured notebook_src_dirs. Use this after cloning a repo where only the .py sources are committed.

Prints a summary of created/updated, unchanged, and skipped files. Raises SystemExit(1) if jupytext reports any errors.

Source code in juplit/tasks.py

def generate_notebooks() -> None:
    """Generate `.ipynb` files from `.py` percent-format files.

    Calls `jupytext --to notebook` on every paired `.py` file found in the
    configured `notebook_src_dirs`. Use this after cloning a repo where only
    the `.py` sources are committed.

    Prints a summary of created/updated, unchanged, and skipped files.
    Raises `SystemExit(1)` if jupytext reports any errors.
    """
    from juplit.artifacts import artifact_py_files, is_artifact

    files = sorted(set(_find_percent_notebook_py_files()) | set(artifact_py_files()))
    if not files:
        print("No percent notebook .py files found.")
        return

    # An artifact whose .ipynb already exists is regenerated with --update, which keeps
    # its outputs; plain --to notebook wipes them, which is the issue #3 data loss.
    _prepare_artifacts(files)
    keep_outputs = [f for f in files if is_artifact(f) and _paired_ipynb(f).exists()]
    plain = [f for f in files if f not in keep_outputs]

    g_plain, e_plain = _run_jupytext(["--to", "notebook"], plain)
    g_keep, e_keep = _run_jupytext(["--to", "notebook", "--update"], keep_outputs)
    groups = {k: g_plain[k] + g_keep[k] for k in g_plain}
    errors = e_plain + e_keep
    _normalize_artifacts(files)
    _save_hashes(files)
    if keep_outputs:
        print(_fmt("nb kept outputs (artifact)", [_key(f, _repo_root()) for f in keep_outputs]))
    _warn_gitignored_artifacts(files)
    if groups["to_ipynb"]:
        print(_fmt("nb created/updated", groups["to_ipynb"]))
    if groups["unchanged"]:
        print(_fmt("nb unchanged", groups["unchanged"]))
    if groups["skipped"]:
        print(_fmt("nb skipped", groups["skipped"]))
    if not any(groups.values()):
        print("Notebooks: nothing to do")
    for err in errors:
        print(f"nb error: {err}")
    stale = _report_stale_artifacts(files)
    if errors or stale:
        raise SystemExit(1)

clean_notebooks(force=False)

Sync then delete .ipynb files, keeping artifact notebooks unless force.

First calls sync_notebooks() to flush any unsaved changes from the .ipynb files back into their paired .py sources, then removes every .ipynb found under notebook_src_dirs. Keeps the working directory clean for AI agents and CI environments that only need the .py sources.

An artifact notebook is the deliverable, not a build product, so it is kept and counted; force=True deletes it too.

Prints a summary of removed and kept files.

Source code in juplit/tasks.py

def clean_notebooks(force: bool = False) -> None:
    """Sync then delete `.ipynb` files, keeping artifact notebooks unless `force`.

    First calls `sync_notebooks()` to flush any unsaved changes from the
    `.ipynb` files back into their paired `.py` sources, then removes every
    `.ipynb` found under `notebook_src_dirs`. Keeps the working directory
    clean for AI agents and CI environments that only need the `.py` sources.

    An artifact notebook is the deliverable, not a build product, so it is kept and
    counted; `force=True` deletes it too.

    Prints a summary of removed and kept files.
    """
    from juplit.artifacts import artifact_py_files

    sync_notebooks()
    artifacts = {_paired_ipynb(f).resolve() for f in artifact_py_files()}
    protected = set() if force else artifacts
    candidates = {f.resolve() for d in _get_src_dirs() for f in d.rglob("*.ipynb")}
    candidates |= {f for f in artifacts if f.exists()}
    removed, kept = [], []
    for f in sorted(candidates):
        if f in protected:
            kept.append(f.name)
            continue
        removed.append(f.name)
        f.unlink()
    if removed:
        print(_fmt("clean removed", sorted(removed)))
    if kept:
        print(_fmt("clean kept artifact notebooks (--force to delete them too)", sorted(kept)))
    if not removed and not kept:
        print("clean: nothing to remove")

    from juplit.kernel import stop_all

    stopped = stop_all()
    if stopped:
        print(_fmt("clean stopped kernels", stopped))

juplit.testing

test() helper — separates exportable logic from inline tests.

test()

True when the calling module is run directly (__main__), or under a pytest run for a module inside that run's own [tool.juplit] notebook_src_dirs.

Use this to gate test code in percent-format notebook files so that tests run interactively (in Jupyter) and under pytest, but never on import — and never for an installed package that happens to be imported by someone else's pytest run.

Example::

# %%
from juplit import test

# %%
def add(a, b):
    return a + b

# %%
if test():
    assert add(1, 2) == 3

Source code in juplit/testing.py

def test() -> bool:
    """True when the calling module is run directly (`__main__`), or under a pytest run
    for a module inside that run's own `[tool.juplit] notebook_src_dirs`.

    Use this to gate test code in percent-format notebook files so that tests
    run interactively (in Jupyter) and under pytest, but never on import — and
    never for an installed package that happens to be imported by someone else's
    pytest run.

    Example::

        # %%
        from juplit import test

        # %%
        def add(a, b):
            return a + b

        # %%
        if test():
            assert add(1, 2) == 3
    """
    caller = sys._getframe(1).f_globals
    if caller.get("__name__") == "__main__":
        return True
    if "pytest" not in sys.modules:
        return False
    file = caller.get("__file__")
    if not file:
        return False
    path = Path(file).resolve()
    return any(path.is_relative_to(root) for root in _test_roots())

Artifact notebooks

juplit.artifacts

Artifact notebooks — pairs whose .ipynb is committed with its outputs.

juplit's default model is "the .py is the source of truth, the .ipynb is a disposable local artifact". That is wrong for one case: a notebook whose executed outputs are the deliverable. A project opts those pairs in by path pattern::

[tool.juplit]
artifact_notebooks = ["experiments/**/*.py"]

For a declared pair, no juplit command destroys the outputs. Committed does not mean untouched, though: every write strips the running state of the run that produced the outputs — execution counts, per-run timestamps, widget state, carriage-return progress bars — so re-running and getting the same results produces no diff.

is_artifact(py_file)

True if this paired .py is declared an artifact notebook.

Source code in juplit/artifacts.py

def is_artifact(py_file: Path) -> bool:
    """True if this paired `.py` is declared an artifact notebook."""
    globs = artifact_globs()
    if not globs:
        return False
    return py_file.resolve() in _matched_paths(_repo_root(), tuple(globs))

artifact_py_files()

Every declared artifact .py, including ones outside notebook_src_dirs.

The globs are additive to the scanned set, so experiments/**/*.py need not also be listed in notebook_src_dirs. Discovery of ordinary pairs is unchanged.

Source code in juplit/artifacts.py

def artifact_py_files() -> list[Path]:
    """Every declared artifact `.py`, including ones outside `notebook_src_dirs`.

    The globs are additive to the scanned set, so `experiments/**/*.py` need not also be
    listed in `notebook_src_dirs`. Discovery of ordinary pairs is unchanged.
    """
    globs = artifact_globs()
    found = {
        f for f in _matched_paths(_repo_root(), tuple(globs)) if _is_paired_notebook(f)
    } if globs else set()
    found.update(f.resolve() for f in _find_percent_notebook_py_files() if is_artifact(f))
    return sorted(found)

cell_state(cell)

Provenance verdict for one cell.

empty — no outputs; nothing to vouch for. clean — stamped, and the stamp matches the current source. stale — stamped, and the stamp disagrees: the outputs describe older source. unverified — has outputs but no stamp: executed outside juplit (a human in Jupyter).

Source code in juplit/artifacts.py

def cell_state(cell: NotebookNode) -> CellState:
    """Provenance verdict for one cell.

    empty       — no outputs; nothing to vouch for.
    clean       — stamped, and the stamp matches the current source.
    stale       — stamped, and the stamp disagrees: the outputs describe older source.
    unverified  — has outputs but no stamp: executed outside juplit (a human in Jupyter).
    """
    if cell.cell_type != "code" or not cell.get("outputs"):
        return "empty"
    stamp = cell.get("metadata", {}).get(STAMP_KEY, {}).get("src_sha256")
    if stamp is None:
        return "unverified"
    return "clean" if stamp == source_sha(cell.source) else "stale"

scan(ipynb)

Group an artifact notebook's cell indices by state. Pure read, never writes.

Source code in juplit/artifacts.py

def scan(ipynb: Path) -> dict[str, list[int]]:
    """Group an artifact notebook's cell indices by state. Pure read, never writes."""
    nb = nbformat.read(ipynb, as_version=4)
    states: dict[str, list[int]] = {"clean": [], "stale": [], "unverified": [], "empty": []}
    for index, cell in enumerate(nb.cells):
        states[cell_state(cell)].append(index)
    return states

stamp(py_file, cells=None, force=False)

Bless outputs juplit did not produce: stamp unverified cells as current.

This is the way out of the unverified warning for a human who ran the notebook in Jupyter. Refuses a stale cell unless force — stamping stale outputs asserts something already known to be false. Returns the indices stamped.

Source code in juplit/artifacts.py

def stamp(py_file: Path, cells: list[int] | None = None, force: bool = False) -> list[int]:
    """Bless outputs juplit did not produce: stamp `unverified` cells as current.

    This is the way out of the `unverified` warning for a human who ran the notebook in
    Jupyter. Refuses a `stale` cell unless `force` — stamping stale outputs asserts
    something already known to be false. Returns the indices stamped.
    """
    ipynb = _paired_ipynb(py_file)
    nb = read_artifact(py_file)
    targets = cells if cells is not None else range(len(nb.cells))
    stamped = []
    for index in targets:
        if index >= len(nb.cells):
            raise ValueError(f"cell {index} is out of range (0..{len(nb.cells) - 1})")
        cell = nb.cells[index]
        state = cell_state(cell)
        if state == "stale" and not force:
            raise ValueError(
                f"cell {index} is STALE; re-execute it, or pass --force to assert "
                "these outputs are current"
            )
        if state in ("unverified", "stale"):
            stamp_cell(cell)
            stamped.append(index)
    if stamped:
        write_artifact(ipynb, nb)
    return stamped

normalize_notebook(py_file)

Scrub a committed artifact on disk and report what it costs.

The file-level counterpart of normalize, which works on an in-memory notebook. Oversized outputs are reported only — never downscaled, never spilled to separate files: a spilled image renders as nothing in GitHub's notebook viewer, which is the one surface where reviewers actually read these notebooks.

Source code in juplit/artifacts.py

def normalize_notebook(py_file: Path) -> dict[str, object]:
    """Scrub a committed artifact on disk and report what it costs.

    The file-level counterpart of `normalize`, which works on an in-memory notebook.
    Oversized outputs are reported only — never downscaled, never spilled to separate
    files: a spilled image renders as nothing in GitHub's notebook viewer, which is the
    one surface where reviewers actually read these notebooks.
    """
    ipynb = _paired_ipynb(py_file)
    nb = read_artifact(py_file)
    changed = normalize(nb) | ensure_filter(nb)
    if changed:
        write_artifact(ipynb, nb)
    return {
        "changed": changed,
        "bytes": ipynb.stat().st_size,
        "oversized": _oversized_outputs(nb),
    }

check_artifacts(strict=False)

The pre-commit / CI guard. Reads committed files only — no kernel, no sidecar.

Fails on: a stale cell, a missing metadata filter, a declared artifact with no committed .ipynb, a gitignored artifact, and a notebook or single output over budget. Warns on unverified cells; strict promotes that warning to a failure.

Raises SystemExit(1) when anything failed.

Source code in juplit/artifacts.py

def check_artifacts(strict: bool = False) -> None:
    """The pre-commit / CI guard. Reads committed files only — no kernel, no sidecar.

    Fails on: a stale cell, a missing metadata filter, a declared artifact with no
    committed `.ipynb`, a gitignored artifact, and a notebook or single output over
    budget. Warns on `unverified` cells; `strict` promotes that warning to a failure.

    Raises `SystemExit(1)` when anything failed.
    """
    from juplit.tasks import _is_gitignored, _key

    py_files = artifact_py_files()
    if not py_files:
        print("check: no artifact notebooks configured")
        return

    root, failures, warnings = _repo_root(), [], []
    for py_file in py_files:
        name = _key(py_file, root)
        ipynb = _paired_ipynb(py_file)
        if not ipynb.exists():
            failures.append(f"{name}: no committed .ipynb — run `juplit nb`")
            continue
        if _is_gitignored(ipynb):
            failures.append(
                f"{name}: its .ipynb is gitignored, so the outputs will never be "
                f"committed — un-ignore it (`!{_key(ipynb, root)}`)"
            )
        nb = nbformat.read(ipynb, as_version=4)
        try:
            nbformat.validate(nb)
        except nbformat.ValidationError as invalid:
            # GitHub's rich notebook diff refuses to render a notebook that fails the
            # schema, so an artifact that does not validate is not reviewable.
            failures.append(f"{name}: .ipynb fails nbformat validation — {invalid.message}")
        py_nb = jupytext.reads(py_file.read_text(), fmt="py:percent")
        for half, node in ((name, py_nb), (_key(ipynb, root), nb)):
            if not has_filter(node.metadata.get("jupytext", {}).get("cell_metadata_filter")):
                failures.append(
                    f"{half}: missing `cell_metadata_filter: {METADATA_FILTER}` — "
                    "provenance stamps leak into the .py without it, and the next sync "
                    "drops them; run `juplit normalize`"
                )
        drift = _source_drift(py_nb, nb)
        if drift is not None:
            failures.append(
                f"{name}: the .py and its .ipynb disagree at cell {drift} — the "
                "committed notebook does not show the committed code (run `juplit sync`)"
            )
        states = scan(ipynb)
        if states["stale"]:
            failures.append(
                f"{name}: cells {_indices(states['stale'])} STALE — outputs predate the "
                f"current .py (`juplit run {name} --stale`, or revert the source edit)"
            )
        if states["unverified"]:
            warnings.append(
                f"{name}: cells {_indices(states['unverified'])} unverified — outputs "
                f"juplit did not produce (`juplit stamp {name}` to vouch for them)"
            )
        size = ipynb.stat().st_size
        if size > max_notebook_bytes():
            failures.append(
                f"{name}: {size:,} bytes exceeds the {max_notebook_bytes():,}-byte "
                "notebook budget"
            )
        for index, output_size in _oversized_outputs(nb):
            failures.append(
                f"{name}: cell {index} has a {output_size:,}-byte output, over the "
                f"{max_output_bytes():,}-byte budget"
            )

    for warning in warnings:
        print(f"check WARNING {warning}")
    for failure in failures:
        print(f"check FAIL {failure}")
    if failures or (strict and warnings):
        raise SystemExit(1)
    print(f"check: {len(py_files)} artifact notebook(s) OK")

add_cell(py_file, source, outputs, index=None, cell_type='code')

Insert one cell into the .py and its captured outputs into the paired .ipynb.

Nothing is re-executed: the outputs written are the ones the caller already looked at. Both halves are built from the same in-memory insert, so they cannot disagree about ordering. Returns the index the cell landed at; index=None appends.

Source code in juplit/artifacts.py

def add_cell(py_file: Path, source: str, outputs: list[NotebookNode],
             index: int | None = None, cell_type: str = "code") -> int:
    """Insert one cell into the `.py` and its captured outputs into the paired `.ipynb`.

    Nothing is re-executed: the outputs written are the ones the caller already looked
    at. Both halves are built from the same in-memory insert, so they cannot disagree
    about ordering. Returns the index the cell landed at; `index=None` appends.
    """
    from juplit.tasks import _save_hashes

    if not is_artifact(py_file):
        raise ValueError(f"{py_file} is not in artifact_notebooks")
    ipynb = _paired_ipynb(py_file)
    py_before, ipynb_before = py_file.read_bytes(), None
    py_nb = jupytext.reads(py_file.read_text(), fmt="py:percent")
    nb = read_artifact(py_file)
    ipynb_before = ipynb.read_bytes()

    drift = _source_drift(py_nb, nb)
    if drift is not None:
        raise ValueError(
            f"{py_file} and its .ipynb are out of sync at cell {drift} — run "
            "`juplit sync` first"
        )
    if index is None:
        index = len(py_nb.cells)
    if not 0 <= index <= len(py_nb.cells):
        raise ValueError(f"index {index} is out of range (0..{len(py_nb.cells)})")

    make = nbformat.v4.new_code_cell if cell_type == "code" else nbformat.v4.new_markdown_cell
    py_nb.cells.insert(index, make(source))
    executed = make(source)
    if cell_type == "code":
        executed.outputs = list(outputs)
        stamp_cell(executed)
    nb.cells.insert(index, executed)

    ensure_filter(py_nb)
    py_file.write_text(jupytext.writes(py_nb, fmt="py:percent"))
    write_artifact(ipynb, nb)

    check = _source_drift(
        jupytext.reads(py_file.read_text(), fmt="py:percent"),
        nbformat.read(ipynb, as_version=4),
    )
    if check is not None:
        py_file.write_bytes(py_before)
        ipynb.write_bytes(ipynb_before)
        raise RuntimeError(
            f"insert left {py_file} and its .ipynb disagreeing at cell {check}; "
            "both files were restored"
        )
    _save_hashes([py_file])
    return index

run_cells(py_file, cells=None, stale_only=False, all_cells=False, name='default', timeout=300.0)

Execute selected cells and SAVE their outputs, freshly stamped, into the .ipynb.

Exactly one selector, and one is required: there is no default because the plausible default — re-running everything — is the expensive one. all_cells is the clean build: the kernel is restarted first so the run starts from nothing.

The restart does not relocate the kernel. A kernel started with --cwd comes back in that directory, so a notebook's relative paths mean the same thing under --all as under --cells. With no session on disk to read, the new kernel lands at the repo root — as it always did — and "fell_back_to" names that directory so a relocated run is visible rather than silent.

Returns {"executed": [...], "failed": [...], "fell_back_to": str | None}; a cell whose output is an error counts as failed but is still written, because that error is what the notebook now shows.

Source code in juplit/artifacts.py

def run_cells(py_file: Path, cells: list[int] | None = None, stale_only: bool = False,
              all_cells: bool = False, name: str = "default",
              timeout: float = 300.0) -> dict[str, object]:
    """Execute selected cells and SAVE their outputs, freshly stamped, into the `.ipynb`.

    Exactly one selector, and one is required: there is no default because the plausible
    default — re-running everything — is the expensive one. `all_cells` is the clean
    build: the kernel is restarted first so the run starts from nothing.

    **The restart does not relocate the kernel.** A kernel started with `--cwd` comes back
    in that directory, so a notebook's relative paths mean the same thing under `--all` as
    under `--cells`. With no session on disk to read, the new kernel lands at the repo
    root — as it always did — and `"fell_back_to"` names that directory so a relocated run
    is visible rather than silent.

    Returns {"executed": [...], "failed": [...], "fell_back_to": str | None}; a cell whose
    output is an error counts as failed but is still written, because that error is what
    the notebook now shows.
    """
    from juplit import kernel as kernel_module   # lazy: `sync` and `check` never pay for it
    from juplit.tasks import _save_hashes

    selectors = [cells is not None, stale_only, all_cells]
    if sum(selectors) != 1:
        raise ValueError(
            "pass exactly one of --cells / --stale / --all "
            "(--all restarts the kernel and re-runs the whole notebook)"
        )

    ipynb = _paired_ipynb(py_file)
    nb = read_artifact(py_file)
    fell_back_to = None
    if stale_only:
        targets = scan(ipynb)["stale"]
    elif all_cells:
        targets = [i for i, cell in enumerate(nb.cells) if cell.cell_type == "code"]
        previous = kernel_module.read_session(name)   # before stop(): it unlinks the record
        kernel_module.stop(name)
        restarted = kernel_module.start(
            name, cwd=Path(previous["cwd"]) if previous else None
        )
        if previous is None:
            fell_back_to = restarted["cwd"]
    else:
        targets = cells

    executed, failed = [], []
    for index in targets:
        if index >= len(nb.cells):
            raise ValueError(f"cell {index} is out of range (0..{len(nb.cells) - 1})")
        cell = nb.cells[index]
        if cell.cell_type != "code":
            raise ValueError(f"cell {index} is markdown; nothing to run")
        cell.outputs = kernel_module.execute(cell.source, name=name, timeout=timeout)
        stamp_cell(cell)
        executed.append(index)
        if any(o.get("output_type") == "error" for o in cell.outputs):
            failed.append(index)

    if executed:
        write_artifact(ipynb, nb)
        _save_hashes([py_file])
    return {"executed": executed, "failed": failed, "fell_back_to": fell_back_to}

juplit.kernel

A Jupyter kernel that outlives the CLI invocation that started it.

Agents call a CLI once per tool use, so "try a cell, look at the result, try again" needs a kernel that survives between calls. There is no Jupyter server here: juplit launches the kernelspec itself and records the connection in .juplit/kernels/, so this works on a login node, in CI and in an ephemeral sandbox.

Two things about the launch are load-bearing, both verified:

• The kernelspec's argv[0] is a bare "python", which inside a virtualenv resolves to the system interpreter — no ipykernel, kernel dead at launch. It has to be rewritten to sys.executable, the same substitution jupyter_client makes internally.
• jupyter_client sets JPY_PARENT_PID to the launching process, and ipykernel then kills the kernel when that process exits. Overriding it to 1 is what makes the kernel detached; without it the kernel does not survive to the first re-attach.

start(name=DEFAULT_NAME, kernel='python3', cwd=None)

Launch a kernel that outlives this process. Reuses a live one of the same name.

Source code in juplit/kernel.py

def start(name: str = DEFAULT_NAME, kernel: str = "python3", cwd: Path | None = None) -> dict:
    """Launch a kernel that outlives this process. Reuses a live one of the same name."""
    if alive(name):
        return read_session(name)

    kernels_dir().mkdir(parents=True, exist_ok=True)
    conn_file = kernels_dir() / f"{name}-connection.json"
    conn_file.write_text(json.dumps(_connection_info(name, conn_file)))
    argv = _kernel_argv(kernel, conn_file)
    log = kernels_dir() / f"{name}.log"

    with open(log, "w") as log_file:
        process = subprocess.Popen(
            argv,
            env={**os.environ, "JPY_PARENT_PID": "1"},
            start_new_session=True,
            cwd=str(cwd or _repo_root()),
            stdin=subprocess.DEVNULL,
            stdout=log_file,
            stderr=subprocess.STDOUT,
        )

    session = {
        "name": name,
        "pid": process.pid,
        "kernel": kernel,
        "cwd": str(cwd or _repo_root()),
        "connection_file": str(conn_file),
    }
    session_path(name).write_text(json.dumps(session, indent=2))
    try:
        _client(session, READY_TIMEOUT).stop_channels()
    except Exception:
        stop(name)
        raise RuntimeError(
            f"kernel {name!r} died before it was ready — see {log}:\n{log.read_text()[-500:]}"
        ) from None
    return session

stop(name=DEFAULT_NAME)

Shut a kernel down and forget it. True if there was a session to stop.

We did not spawn it in this process, so there is no KernelManager to ask: the pid in the session file and the control channel are all we have.

Source code in juplit/kernel.py

def stop(name: str = DEFAULT_NAME) -> bool:
    """Shut a kernel down and forget it. True if there was a session to stop.

    We did not spawn it in this process, so there is no KernelManager to ask: the pid in
    the session file and the control channel are all we have.
    """
    session = read_session(name)
    if session is None:
        return False
    try:
        client = _client(session, LIVENESS_TIMEOUT)
        client.shutdown()
        client.stop_channels()
    except Exception:
        try:
            os.kill(session["pid"], signal.SIGTERM)
        except OSError:
            pass

    for path in (session_path(name), Path(session["connection_file"])):
        path.unlink(missing_ok=True)
    for socket_file in kernels_dir().glob(f"{name}-*"):
        if not socket_file.name.endswith(".json"):
            socket_file.unlink(missing_ok=True)
    return True

status()

One row per recorded session: name, pid, cwd, kernel, alive.

Source code in juplit/kernel.py

def status() -> list[dict]:
    """One row per recorded session: name, pid, cwd, kernel, alive."""
    if not kernels_dir().exists():
        return []
    rows = []
    for path in sorted(kernels_dir().glob("*.json")):
        if path.name.endswith("-connection.json"):
            continue
        session = read_session(path.stem)
        if session is not None:
            rows.append(session | {"alive": alive(session["name"])})
    return rows

execute(code, name=DEFAULT_NAME, timeout=DEFAULT_TIMEOUT)

Run code on the session kernel and return its outputs. Touches no file.

A failed attempt therefore exists only in the kernel's history and in this return value — never in the notebook.

Source code in juplit/kernel.py

def execute(code: str, name: str = DEFAULT_NAME,
            timeout: float = DEFAULT_TIMEOUT) -> list[NotebookNode]:
    """Run code on the session kernel and return its outputs. Touches no file.

    A failed attempt therefore exists only in the kernel's history and in this return
    value — never in the notebook.
    """
    session = read_session(name)
    if session is None:
        raise RuntimeError(f"no kernel {name!r} — run `juplit kernel start`")
    try:
        os.kill(session["pid"], 0)
    except OSError:
        raise RuntimeError(
            f"kernel {name!r} is gone (pid {session['pid']}) — run `juplit kernel start`"
        ) from None

    try:
        client = _client(session, LIVENESS_TIMEOUT)
    except Exception:
        raise RuntimeError(
            f"kernel {name!r} is wedged — run `juplit kernel stop --name {name}`"
        ) from None

    outputs: list[NotebookNode] = []
    msg_id = client.execute(code)
    try:
        while True:
            message = client.get_iopub_msg(timeout=timeout)
            if message["parent_header"].get("msg_id") != msg_id:
                continue
            output = _to_output(message)
            if output is not None:
                outputs.append(output)
            if (message["msg_type"] == "status"
                    and message["content"]["execution_state"] == "idle"):
                return outputs
    except Exception as exc:
        if type(exc).__name__ != "Empty":                # queue.Empty means the timeout hit
            raise
        _interrupt(session)
        raise RuntimeError(
            f"execution exceeded {timeout}s and was interrupted; "
            f"partial output: {[o.get('output_type') for o in outputs]}"
        ) from None
    finally:
        client.stop_channels()

juplit.view

Reading a notebook without paying for it.

A plot-heavy .ipynb is megabytes of base64. An agent that reads one into context has spent its budget on data it cannot interpret. These renderers never emit a payload: an image output becomes image/png 640x480 82KB, a 10 000-line stream becomes its first and last lines, and only text is ever shown in full.

cells_table(ipynb)

The index: one line per cell, headings and digests, never any source or base64.

[00] md     6L   # Ablation study
[05] md     2L   ## Results
[06] code  12L   clean        out: stream 1.2KB 14L 'Fitting fold 3…'
[07] code   4L   STALE        out: text/plain 3L '[1200 rows x 8 columns]'

Every column left of the digest is fixed-width, and the digest — the one field with no common structure — comes last, so clean / STALE land in the same place on every line instead of after a string whose length depends on the output.

Source code in juplit/view.py

def cells_table(ipynb: Path) -> str:
    """The index: one line per cell, headings and digests, never any source or base64.

        [00] md     6L   # Ablation study
        [05] md     2L   ## Results
        [06] code  12L   clean        out: stream 1.2KB 14L 'Fitting fold 3…'
        [07] code   4L   STALE        out: text/plain 3L '[1200 rows x 8 columns]'

    Every column left of the digest is fixed-width, and the digest — the one field with
    no common structure — comes last, so `clean` / `STALE` land in the same place on
    every line instead of after a string whose length depends on the output.
    """
    from juplit.artifacts import cell_state

    nb = nbformat.read(ipynb, as_version=4)
    rows = []
    for index, cell in enumerate(nb.cells):
        kind = "code" if cell.cell_type == "code" else "md"
        length = f"{len(cell.source.splitlines())}L"
        row = f"[{index:02d}] {kind:<4} {length:>4}"
        if cell.cell_type == "markdown":
            heading = markdown_heading(cell.source)
            if heading:
                row = f"{row}   {heading[:80]}"
        if cell.get("outputs"):
            state = cell_state(cell)
            label = state.upper() if state == "stale" else state
            digests = ", ".join(output_digest(o) for o in cell.outputs)
            row = f"{row}   {label:<11}out: {digests}"
        rows.append(row)
    return "\n".join(rows)

view_cells(py_file, cells=None, full=False)

Source of the named cells, each followed by its outputs. No range means all.

Both halves are introduced by the same banner rule::

── [03] code ───────────────────────────────────────────────────
for arm, scores in ARMS.items():
    print(f"{arm:<10} mean {sum(scores) / len(scores):.2f}")
── [03] output ─────────────────────────────────────────────────
baseline   mean 0.41

Source comes from the .py — the source of truth — and outputs from the paired .ipynb, so this works on an ordinary pair with no notebook on disk: it prints the source and says the notebook has not been generated.

Source code in juplit/view.py

def view_cells(py_file: Path, cells: list[int] | None = None, full: bool = False) -> str:
    """Source of the named cells, each followed by its outputs. No range means all.

    Both halves are introduced by the same banner rule::

        ── [03] code ───────────────────────────────────────────────────
        for arm, scores in ARMS.items():
            print(f"{arm:<10} mean {sum(scores) / len(scores):.2f}")
        ── [03] output ─────────────────────────────────────────────────
        baseline   mean 0.41

    Source comes from the `.py` — the source of truth — and outputs from the paired
    `.ipynb`, so this works on an ordinary pair with no notebook on disk: it prints the
    source and says the notebook has not been generated.
    """
    import jupytext

    nb = jupytext.reads(py_file.read_text(), fmt="py:percent")
    ipynb = py_file.with_suffix(".ipynb")
    outputs: list[list[NotebookNode]] = []
    if ipynb.exists():
        executed = nbformat.read(ipynb, as_version=4)
        outputs = [cell.get("outputs", []) for cell in executed.cells]

    wanted = cells if cells is not None else range(len(nb.cells))
    blocks = []
    for index in wanted:
        if index >= len(nb.cells):
            raise ValueError(f"cell {index} is out of range (0..{len(nb.cells) - 1})")
        cell = nb.cells[index]
        kind = "code" if cell.cell_type == "code" else "markdown"
        block = [banner(index, kind), cell.source.rstrip()]
        for output in outputs[index] if index < len(outputs) else []:
            rendered = (truncate_output(output, max_lines=10_000, max_chars=200_000)
                        if full else truncate_output(output))
            block += [banner(index, "output"), rendered]
        blocks.append("\n".join(block))
    if not ipynb.exists():
        blocks.append(f"(no {ipynb.name} on disk — source only; run `juplit nb` to generate it)")
    return "\n\n".join(blocks)

output_digest(output)

A one-line, base64-free summary of a single output.

Source code in juplit/view.py

def output_digest(output: NotebookNode) -> str:
    """A one-line, base64-free summary of a single output."""
    kind = output.get("output_type")
    if kind == "stream":
        text = output.get("text", "")
        return (f"stream({output.get('name', 'stdout')}) {_human_bytes(len(text))} "
                f"{len(text.splitlines())}L {_first_and_last(text)}").strip()
    if kind == "error":
        frames = len(output.get("traceback", []))
        return f"error {output.get('ename')}: {output.get('evalue')} ({frames} frames)"

    parts = []
    for mime, payload in output.get("data", {}).items():
        text = payload if isinstance(payload, str) else str(payload)
        if mime.startswith(BINARY_PREFIXES):
            size = _human_bytes(len(text) * 3 // 4)
            dims = _png_dimensions(text) if mime == "image/png" else None
            parts.append(f"{mime} {dims[0]}x{dims[1]} {size}" if dims else f"{mime} {size}")
        elif mime == "text/plain":
            parts.append(f"text/plain {len(text.splitlines())}L {_first_and_last(text)}")
        else:
            parts.append(f"{mime} {_human_bytes(len(text))}")
    return ", ".join(parts) or kind