Metadata-Version: 2.5
Name: agentknit
Version: 0.43.0
Summary: Build reliable coding agents for any OpenAI-compatible /chat/completions endpoint—with spec-driven tools, observability, cache verification, context compaction, and sandboxing. Less glue code, fewer crossed fingers.
Author: agentknit contributors
License: MIT
License-File: LICENSE
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Requires-Python: >=3.11
Requires-Dist: requests>=2.28
Provides-Extra: keyring
Requires-Dist: keyring>=25; extra == 'keyring'
Description-Content-Type: text/markdown

# agentknit

Coding agent framework for any `/chat/completion` endpoint.

Features:
* Reads a [JSON spec](specification.md) to dispatches tool calls (read_file, write_file, execute_bash, …) to tool implementations.
* Grammar-constrained custom tools for endpoint-side constrained decoding
* Fail-closed strict cache-proof mode for saving your $$$$$
* Rich event system (`tool_call`, `content_delta`, `usage`, …) for logging/TUI/dashboards
* Automatic context compaction keeps long sessions inside the token budget
* Bubblewrap-sandboxed tool execution for untrusted replay workloads
* `rtk` integration cuts shell tool output tokens by 60–90%

## Install

```
pip install agentknit
```

### Programmatic

```python
from agentknit import load_specification, run_task

schema = load_specification("qwen3-8b.json", "https://openrouter.ai/api/v1")
result = run_task(schema, "List the files in /tmp")
print(result.final_reply)
```

To load a pre-written spec file directly (no name-based lookup, no probing), pass
`spec_path=`:

```python
schema = load_specification(
    "deepseek-v4-flash-free",
    "run:///home/user/bin/completions.py",
    spec_path="/home/user/.config/agentknit/agent_spec.json",
)
```

### Quick scripts with direct tools

For small agents, use `run_agent()` to provide the model connection and
`Tool` definitions directly. It builds the internal schema and registers the
tool callables for you:

```python
from pathlib import Path

from agentknit import Tool, run_agent


def list_files(path: str) -> tuple[str, dict]:
    return ", ".join(p.name for p in Path(path).iterdir()), {"result": "ok"}


result = run_agent(
    task="List the files in /tmp",
    model="deepseek-v4-flash-free",
    endpoint="https://opencode.ai/zen/v1",
    auth="opencode-github-copilot",
    tools=[Tool("list_files", "List a directory", list_files)],
)
print(result.final_reply)
```

### Injecting a custom client

All entry points — `run_task`, `run_agent`, `run` and `run_repl`/`run_async_repl` — accept
an optional `client=`. Pass your own `OpenAI`- or `SubprocessOpenAI`-compatible
object (sandbox client, wrapper, instrumented subclass) instead of the one
`create_client` builds from the spec:

```python
from agentknit import load_specification, run_repl
from mybridge import GrammarOpenAI   # subclass of SubprocessOpenAI

client = GrammarOpenAI("./completions.sh")
agentknit.run_repl(load_specification("run://grammar"), client=client)
```

The same client can be reused for `init_session` + `run_turn` one-shot calls
in the same script — no monkey-patching of `create_client`.


## Strict Cache Proof

`agentknit` now runs in strict cache-proof mode by default.

After the first LLM call in a session, every later call must return explicit
server-side cache accounting with a nonzero cache hit. If the provider does
not expose cache-proof fields, or reports `cached_tokens = 0`, the run aborts.

This is fail-closed by design: models that cannot prove cache reuse should be
treated as unsupported for cache-sensitive workloads.

The usage layer normalizes several provider response shapes into one check,
including:

- `usage.prompt_tokens_details.cached_tokens`
- `usage.cache_read_input_tokens`
- `usage.cache_read_tokens`
- cache-write fields such as `cache_creation_input_tokens` and `cache_write_tokens`

Programmatic calls accept `strict_cache_proof=True` by default:

```python
result = run_task(schema, "List the files in /tmp", strict_cache_proof=True)
```

CLI usage is also strict by default. To opt out explicitly:

```bash
agent-probe <model> --no-strict-cache-proof
```

#### Provider minimum cacheable prompt size

Some providers cache nothing below a documented minimum prompt size —
e.g. Anthropic Claude Haiku requires roughly 4096 input tokens before any
caching kicks in, while GPT-5.6-class models cache from roughly 1024 tokens.
A short prompt that legitimately misses the cache would otherwise trip
strict cache-proof mode and abort the run with `CacheProofError`.

Set `min_cacheable_tokens` (prompt-token floor) to tell agentknit that a
zero-cache-hit response is *expected*, not a failure, whenever the current
call's prompt is below that floor:

```python
result = run_task(schema, "Hello", min_cacheable_tokens=4096)
```

Or in the agent spec JSON:

```json
{
  "model": "...",
  "min_cacheable_tokens": 4096
}
```

Or from the CLI:

```bash
agent-probe <model> --min-cacheable-tokens 4096
```

Defaults to `0` (no minimum) — any zero-cache-hit call after the first is
still treated as a genuine cache miss unless you configure this.

### Defining tools with `Tool` & `build_tool_spec`

Declare tools using the `Tool` dataclass and convert them into the schema/dispatch
pair that the agent loop expects with `build_tool_spec`:

```python
from agentknit import Tool, build_tool_spec, register_tools_in_library
from agentknit.tool_library import TOOL_LIBRARY


def t_read(path: str) -> tuple[str, dict]:
    """Read and return the contents of a file."""
    return Path(path).read_text(), {"result": "ok"}


def t_write(path: str, content: str) -> tuple[str, dict]:
    """Write content to a file."""
    Path(path).write_text(content)
    return f"wrote {len(content)} bytes", {"result": "ok"}


tools = [
    Tool("read_file", "Read a file", t_read,
         parameters={"type": "object",
                     "properties": {"path": {"type": "string"}},
                     "required": ["path"]}),
    Tool("write_file", "Write a file", t_write,
         parameters={"type": "object",
                     "properties": {"path": {"type": "string"},
                                    "content": {"type": "string"}},
                     "required": ["path", "content"]}),
]

# Build the OpenAI-compatible schema and dispatch dict
schema, dispatch = build_tool_spec(tools)

# Register the callables so the dispatch loop can find them
register_tools_in_library(tools)

# Now schema and dispatch can be used with init_session / run_task
```

#### Custom (grammar-constrained) tools

`Tool(custom_format=...)` declares an OpenAI **custom tool**: the argument is
raw text, optionally constrained by a grammar for endpoint-side constrained
decoding — no JSON Schema is synthesized:

```python
tools = [Tool(
    "apply_patch",
    "Apply a patch that adds, updates, moves or deletes files.",
    t_apply_patch,                # fn(input: str) -> (str, dict)
    custom_format={"type": "grammar", "syntax": "lark", "definition": GRAMMAR},
)]
schema, dispatch = build_tool_spec(tools)
# schema[0] == {"type": "custom", "name": "apply_patch", ..., "format": {...}}
```

The model's raw text arrives as the single `input` keyword argument.

The `Tool` dataclass also supports `param_map` for translating model-facing
argument names to Python keyword argument names:

```python
tools = [
    Tool("read_file", "Read a file", t_read,
         param_map={"file_path": "path"}),
]
```

### Tool library

The framework ships with a built-in set of tools (`read_file`, `write_file`,
`str_replace`, `exec_shell`). The previous name `execute_shell_command` is
still accepted as an alias. `str_replace` replaces the first occurrence of
`old_str` by default; pass `replace_all: true` to replace every occurrence.

### Background shell tools (`nohup` / `nohup_query` / `nohup_wait`)

`agentknit.async_toolkit` provides bounded background execution on top of the
low-level `t_execute_async` / `t_query_exec` primitives: stdout/stderr are
captured to files, stdin is exposed as a FIFO, and execution is wrapped in
`timeout(1)`. The tool definitions and implementations both live there — add
them to a spec with one call:

```python
import agentknit
from agentknit.async_toolkit import enable_nohup

schema = agentknit.load_specification(MODEL, ENDPOINT)
enable_nohup(schema)   # appends nohup + nohup_query + nohup_wait specs and dispatch entries
```

It is idempotent and supports both schema shapes (`tools` list or pre-built
`tool_dispatch`). The default bound is 10 minutes (`NOHUP_TIMEOUT_MIN`).

Async tools always come with three:

| Tool | Role |
|---|---|
| `nohup` | Start a shell command in the background, bounded by `timeout` minutes (default 10). Returns `tool_exec_id`, the `pid`, and the local stdin (FIFO)/stdout/stderr paths. `wait_before_s` starts the command N seconds later (the `timeout` budget counts from the start, not from the delay). |
| `nohup_query` | Poll one execution by `tool_exec_id`; inlines stdout/stderr when both fit in 4 KiB, otherwise reports file sizes. |
| `nohup_wait` | `nohup_wait(tool_exec_id, howmuch, unit)` waits for **one** background execution instead of busy-polling; `howmuch` × `unit` (`s`/`m`/`h`/`d`) is an optional budget. Returns as soon as the exec completes (returncode, output paths, last lines of stdout/stderr, inline output when small). If the budget expires first, it reports `completed: false` plus the CPU and I/O activity of the still-running process so you can tell progress from a hang. Execs finishing meanwhile are listed under `also_completed`. |

`nohup_wait` is the tool to call right after `nohup`: it blocks until the given
`tool_exec_id` finishes (or the optional `howmuch` budget expires) and returns
the result, so short commands need no `nohup_query` round trip. Waits are
capped at `WAIT_FOR_MAX_SECONDS` (3600) per call; split longer waits into
several calls.

To run a command later — e.g. re-check CI in 5 minutes — pass
`wait_before_s=300`: the `nohup` call returns immediately with the
`tool_exec_id`, the command starts after the delay, the `timeout(1)` budget
applies only once it runs, and `nohup_wait` reports it as usual.

Two consecutive `nohup_query` calls for the same still-running
`tool_exec_id` get a response pointing at
`nohup_wait(tool_exec_id, howmuch, unit)`, which returns as soon as the
execution finishes. The redirect lifts as
soon as anything else happens — another exec is polled, a new command is
started, or the execution completes.

### Sandboxed tool execution (Linux)

Direct local tool dispatch remains the default. For untrusted replay workloads,
pass a `BubblewrapToolExecutor` to keep file and shell tools in an isolated
workspace while the controller retains the model credential and network access:

```python
from pathlib import Path
from agentknit import BubblewrapToolExecutor, SandboxPolicy, run_task

executor = BubblewrapToolExecutor(SandboxPolicy(
    workspace=Path("/tmp/replay-worktree"),
    network="none",
    environment={"PATH": "/usr/bin:/bin"},
))
result = run_task(schema, task, tool_executor=executor)
```

The Bubblewrap executor supports the built-in file tools and synchronous shell
commands. It rejects custom Python and asynchronous tools unless they provide a
sandbox adapter; it never falls back to local execution. Paths are restricted
to the workspace and the selected sandbox policy is recorded in the session log.


## Event System

agentknit emits events during agent execution so you can build logging
frameworks, GUI/TUI integrations, streaming dashboards, or custom monitoring
on top of the framework.

### Subscribing to events

Use `subscribe(session, event_type, handler)` to register a handler for a
specific event type:

```python
from agentknit import subscribe, init_session, run_turn, create_client

schema = load_specification("qwen/qwen3-8b", "https://openrouter.ai/api/v1", False)
client = create_client(schema)
session = init_session(schema)

# Log every tool call
subscribe(session, "tool_call", lambda event_type, data: print(f"[tool] {data['name']}"))

# Stream content deltas in real-time
subscribe(session, "content_delta", lambda event_type, data: print(data.get("text", ""), end=""))

# Track token usage
subscribe(session, "usage", lambda event_type, data: print(f"[tokens] {data}"))

# React to errors
subscribe(session, "error", lambda event_type, data: print(f"[error] {data['text']}"))
```

`init_session()` returns a `Session` — a `TypedDict`, i.e. a plain `dict` at
runtime, with typed, IDE-completable keys (`session["messages"]`,
`session["usage_totals"]`, …). Import it from the package when you want the
annotations in your own code:

```python
from agentknit import Session, init_session

session: Session = init_session(schema)
```

The `on` function is a convenience alias for `subscribe`:
```python
from agentknit import on

on(session, "tool_call", my_handler)
```

Multiple handlers can be registered for the same event type; they are called
in registration order.

### Unsubscribing

```python
from agentknit import unsubscribe

unsubscribe(session, "tool_call", my_handler)
```

### Generic handler

The lower-level `EventCallback` can be passed to `init_session()` via the
`on_event` keyword and receives *all* events:

```python
from agentknit import EventCallback

def my_handler(event_type: str, data: dict) -> None:
    print(f"[{event_type}] {data.get('fmt', data)}")

session = init_session(schema, on_event=my_handler)
```

Per-event-type handlers registered via `subscribe` are called *before* the
generic `on_event` handler.

### Full list of event types

| Event type | When it fires | Data keys |
|---|---|---|
| `tool_call` | Before dispatching a tool | `name`, `args`, `fmt` |
| `tool_result` | After receiving tool result | `name`, `result`, `streamed`, `files`, `diff_summary`, `fmt` |
| `content_delta` | Streaming text chunk from the model | `text`, `first`, `no_newline`, `fmt` |
| `reasoning_delta` | Streaming reasoning trace | `text`, `first`, `no_newline`, `fmt` |
| `content_stream_end` | End of a streaming content sequence | `no_newline`, `fmt` |
| `reasoning_stream_end` | End of a streaming reasoning sequence | `no_newline`, `fmt` |
| `usage` | Per-turn token usage report | `prompt`, `completion`, `total`, `cached`, `cache_write`, `fmt` |
| `session_usage` | Cumulative session usage at final answer | `prompt`, `completion`, `total`, `cached`, `cache_write`, `fmt` |
| `error` | API or dispatch error | `text`, `error_class`, `http_status`, `error_code`, `error_message`, `elapsed_s`, `adapter`, `fmt` |
| `final_answer` | Agent produces its final reply | `text`, `fmt` |
| `token_limit` | Token budget exceeded | `used`, `limit`, `fmt` |
| `session_resumed` | Session history was loaded from disk | `session_id`, `messages_loaded`, `fmt` |
| `provider_pinned` | OpenRouter provider was locked for the session | `provider`, `fmt` |
| `compaction` | Context was compacted into a summary | `summary`, `compacted_turns`, `fmt` |
| `cache_cold` | Resumed turn missed the (expired) prefix cache | `age`, `fmt` |
| `journal_recovered` | A resumed session was rebuilt from the durable journal | `entries_replayed`, `messages_loaded`, `pending`, `mid_turn`, `fmt` |
| `rate_limit_wait` | Before sleeping through a retryable HTTP 429 | `delay`, `resume_at`, `fmt` |

Every event data dict includes a `"fmt"` key containing a pre-formatted ANSI
string suitable for direct printing to a terminal — this is what the default
handler uses.  Custom handlers may ignore `"fmt"` and use the other keys
instead.

The `tool_result` event includes additional metadata for file-writing tools:

- **`files`** — a list of file paths that were created or modified by the tool
  call (e.g. `["src/main.py"]`).  `None` for tools that don't touch files.
- **`diff_summary`** — a dict with `path`, `added` (lines added), and `removed`
  (lines removed) so consumers can display summaries like `+5 -2 src/main.py`
  without re-reading the file.  `None` for non-file tools.

Example::

    subscribe(session, "tool_result", lambda et, data: print(
        f"Files changed: {data.get('files')}  "
        f"Diff: {data.get('diff_summary')}"
    ))

## Context Compaction

Long sessions automatically compact when the prompt token budget is exceeded.
The oldest portion of the conversation is summarized by the model into a
continuation-oriented summary that preserves coding state (objectives, files
touched, errors, next steps). The summary replaces the compacted prefix, while
the most recent turns remain in raw form.

Compaction is **enabled by default** and configured via the agent spec or
programmatic arguments:

```python
from agentknit import run_task

result = run_task(
    schema,
    "Implement feature X",
    compaction_enabled=True,
    compaction_trigger_tokens=100_000,   # trigger when prompt tokens reach this
    compaction_target_tokens=20_000,     # max tokens for the summary call
    compaction_keep_last_turns=2,        # raw turns to keep after compaction
)
```

Or in the agent spec JSON:

```json
{
  "model": "...",
  "compaction_enabled": true,
  "compaction_trigger_tokens": 100000,
  "compaction_target_tokens": 20000,
  "compaction_keep_last_turns": 2
}
```

The summary message is tagged with `"compacted_summary": true` so consumers
can distinguish compacted state from raw conversation turns. Compaction events
are emitted as `"compaction"` events and logged to the session trace.

## Durable Recovery

Snapshots alone only persist the conversation at *turn boundaries* — a crash
mid-turn loses the whole turn from the transcript, including tool calls whose
side effects (file writes, shell commands, deploys) already happened. Resuming
from the snapshot then makes the model blindly re-run those tools.

By default (`durable=True`) agentknit keeps an append-only, fsync-per-record
write-ahead journal next to the snapshot:

```
~/.local/share/agent_probe/<model>/<session_id>_journal.jsonl
```

Every state transition inside a turn is journaled *as it happens*:

| Record | Written | Meaning on recovery |
|---|---|---|
| `message` | when a message joins the history | the conversation is rebuildable past the last snapshot |
| `tool_start` | **before** a tool executes | crash before `tool_end` → side effects unknown |
| `tool_end` | **after** the tool returns | result is known; never re-run it |
| `turn_start` / `turn_end` | around each turn | `turn_start` without `turn_end` → crashed mid-turn |
| `reset_messages` | compaction, `/clear` | history replacement is replayed, not lost |

On resume (`--session <id>`) the journal is replayed and takes precedence over
a stale snapshot:

- messages the snapshot never saw are recovered;
- a tool that finished but whose result the model never saw is **not re-run** —
  the recorded result is re-injected into the conversation as a recovery note;
- a tool that was in flight when the process died has *unknown* side effects —
  a recovery note tells the model to verify state (inspect files, read-only
  checks) before re-running anything.

A torn tail write (crash mid-line) is ignored: an incomplete record was never
acknowledged.

Disable with `durable=False` (programmatic), `"durable": false` in the agent
spec JSON, or `--no-durable` on the CLI:

```python
from agentknit import run_task

result = run_task(schema, task, durable=False)
```

### Explicit session directories and durable sinks

Pass `session_dir` to keep one session's journal, snapshot, and event log in
an explicit folder. This enables complete lifecycle capture: initial prompts,
messages, model requests and responses, streaming frames, tool boundaries, and
runtime events are committed before they are sent to a model, tool, renderer,
or subscriber.

```python
from agentknit import run_task

result = run_task(schema, task, session_dir="/srv/agent-sessions/job-42")
```

`durable_sink` accepts an object with a synchronous `append(record)` method.
It receives the same ordered records after the built-in filesystem journal has
been fsync'd and before downstream consumers run. Raising from `append` stops
the producing operation, so queueing a background write does not satisfy the
contract.

```python
class AuditSink:
    def append(self, record: dict) -> None:
        store_and_fsync(record)

result = run_task(schema, task,
                  session_dir="/srv/agent-sessions/job-42",
                  durable_sink=AuditSink())
```

A `journal_recovered` event is emitted whenever a resume rebuilt state from
the journal.

## Resuming sessions

A session is resumed with ``--session <id>`` (CLI) or ``session_id=`` (SDK).
Resume always continues on the endpoint the session was **created** on, not on
whatever the current `--endpoint` / CLI default resolves to: the endpoint is
recovered from the session's append-only logs (the earliest `session_start`
record), which also protects against a snapshot overwritten by an earlier
resume against a different provider. The recorded key source
(`keyring_service`+`keyring_username` or `key_env`) is restored with it, and
key sources the resumed session never used are dropped. When the resolved
endpoint differs from the requested one, a one-line notice is printed.

### Resuming with another provider/model

`agentknit <other-model> --session <id>` deliberately switches provider. The
session file is **copied** into the new model's directory and re-stamped —
`metadata.model` / `endpoint` / `auth` describe the new provider, and the old
one is kept in `metadata.ported_from` — so the transcript continues on the
endpoint/key of the model you named, and a later resume of that session binds
to the new provider. The original file is left untouched as the record of
where the history came from.

## rtk Integration (optional token savings)

[rtk](https://github.com/rtk-ai/rtk) is a CLI proxy that rewrites shell
command output for 60–90% token savings. When installed, you can opt in by
calling `enable_rtk_rewrite()` once before `run_task()`:

```python
from agentknit import enable_rtk_rewrite, run_task

enable_rtk_rewrite()   # no-op if rtk is not in PATH

result = run_task(schema, "List the files in /tmp")
```

This patches `t_run` and `t_execute_async` in the tool library so every shell
command passes through `rtk rewrite` before execution. It is off by default.
