agentusage

package
v0.16.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 27, 2026 License: MIT Imports: 25 Imported by: 1

Documentation

Overview

Package agentusage reports the token usage AI coding agents record on disk.

Typical use: LoadDefinitions, Discover running agents, Watch each process (or Process.Watch), then Poll or Run for Sample values. A Sample is the total since the watcher attached, so a caller reporting events takes the growth from Sample.Delta. EnableOpenCodeDB opts into opencode's machine-wide SQLite store; crush is read whenever the sqlite build tag is on.

An agent is read one of two ways. Transcript agents (claude, qwen, dsh, clanker, copilot, codex, kimi) appear in the adapters table, each naming where its logs live under a working directory and how one line becomes a Sample. RegisterSpec adds one this package does not ship with, and LoadDefinitions reads the same declaration from a JSON file, DefinitionsPath being the default location.

Database agents are registered as sources instead: crush is built in, opencode is added by EnableOpenCodeDB because its store is machine-wide and the operator opts into it.

Discover finds the agent processes running now, and Watch reads the transcripts of the one working in a directory, so a caller can take a Sample on an interval without knowing which agent is underneath.

The agent registry is process-wide, since one process reports one set of agents. RegisterSpec and UnregisterSpec add and remove an adapter, and LoadDefinitions and ResetDefinitions do the same for a definitions file, so a program that teaches this package an agent can take it back out.

The registry calls and Watch are safe to call from several goroutines at once, and a Watcher is safe to use while its Run is going: Poll takes the read lock a running Run also takes, so a final read after the agent exits is a call like any other rather than a race. A definitions file reloaded after a watcher started reaches it on the next poll, and a source withdrawn by EnableOpenCodeDB(false) stops being read the same way; an adapter installed by RegisterSpec is fixed for the life of the process, and a watcher keeps the one it attached with.

Every method on a nil *Watcher is safe to call on the result of Watch: Tool and Dir report the empty string, Err matches ErrUnsupportedTool, Poll and Sample report the zero Sample, Run returns at once, and SetNow does nothing. A caller skipping the agents it cannot read therefore asks Err, which is what names that case, rather than testing the pointer: 

if w := agentusage.Watch(tool, dir, time.Now()); w.Err() != nil {
	// no readable usage for this agent
}

The crush and opencode sources need a SQLite driver, so they exist only under the sqlite build tag. Without it the package still compiles, and Supported reports those agents unreadable.

Agents differ in what they print to stdout: some report token usage as they stream, some only at exit, some never. They agree on something else, though, which is that they keep a structured session transcript, and that transcript carries per-message usage with timestamps. Tailing it gives a live rate without root, without intercepting anyone's network traffic, and without asking the agent to behave differently.

The design constraints that shape everything here:

  • Only count usage after the watcher attached. Session transcripts persist across runs, so the watcher records where each file ended when it attached and reads only what is appended after that. Database-backed agents (opencode, crush) use the since argument the same way; file transcripts are always tailed from their attach-time end.
  • Attribute the transcript to the right process. Each adapter ties its files to a working directory: recorded per record, read from the session header, or implicit because the log lives inside the project directory itself (clanker), so the cwd is the key.
  • Never invent a number. An agent whose transcript cannot be found, parsed, or attributed simply reports nothing, and the dashboard shows no rate.
Example 
package main

import (
	"context"
	"fmt"
	"sync"
	"time"

	"github.com/maci0/toktop/agentusage"
)

func main() {
	// Typical integration: load extra agent definitions, discover running
	// agents, and tail the one that is working in this directory.
	if err := agentusage.LoadDefinitions(agentusage.DefinitionsPath()); err != nil {
		fmt.Println(err) // malformed or unreadable; a missing file is not an error
	}
	// EnableOpenCodeDB reports whether this build can read opencode's store.
	// Call it before Watch; a false return is a build without -tags sqlite.

	if !agentusage.EnableOpenCodeDB(true) {
		fmt.Println("opencode: build without -tags sqlite")
	}

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	var wg sync.WaitGroup
	for _, p := range agentusage.Discover() {
		w := p.Watch(time.Now())
		if w.Err() != nil {
			continue // the agent keeps nothing this package can read
		}
		wg.Go(func() {
			// Run hands over a running total, so a caller that emits events
			// reports the growth from its previous sample. A delta reporting no
			// growth is the sample to measure from next time, whether the agent
			// was quiet or its transcript was rewritten under the watcher.
			var prev agentusage.Sample
			w.Run(ctx, agentusage.DefaultPollInterval, func(cur agentusage.Sample) {
				if d, ok := cur.Delta(prev); ok {
					fmt.Printf("%s pid %d: %d output, %d prompt\n", p.Tool, p.PID, d.Output, d.Input)
				}
				prev = cur
			})
		})
	}
	wg.Wait()
}

Index

Examples

Constants

View Source 
const DefaultPollInterval = 250 * time.Millisecond

DefaultPollInterval is how often a transcript is re-read, and what Watcher.Run polls at when its interval is not positive. It bounds how stale a live rate can be, so it is tighter than the directory rescan: reading one growing file is cheap, walking a store of thousands is not.

It is exported so a caller naming an interval explicitly passes the package's own default rather than a number copied out of this documentation, which drifts the moment either side changes.

Variables

View Source 
var (
	// ErrEmptyTool is returned by RegisterSpec when the agent name is blank.
	ErrEmptyTool = errors.New("usage spec needs an agent name")
	// ErrNoRoots is returned by RegisterSpec when the spec names no transcript
	// directories. errors.Is matches it through the formatted error that
	// includes the agent name.
	ErrNoRoots = errors.New("usage spec has no roots")
	// ErrUnsupportedTool is what Watcher.Err reports for the nil watcher Watch
	// returns: the agent keeps nothing this package can read, because no source
	// is registered for it, no definition names its transcripts, or its
	// definition names none. It is a fact about the agent, not a failure, so a
	// caller that has nothing to display treats it like any other empty
	// reading.
	ErrUnsupportedTool = errors.New("agent has no readable usage source")
)
View Source 
var ErrCollidingDefinitions = errors.New("agent names collide after NFC normalization")

ErrCollidingDefinitions marks an agents.json holding two names that NFC reduces to one canonical key, an overlap the per-name checks in LoadDefinitions cannot see. It is wrapped inside an ErrInvalidDefinitions error that names the file and both spellings, so a caller can tell a file whose JSON is wrong from one whose agent names are, and say which.

View Source 
var ErrInvalidDefinitions = errors.New("malformed agent definitions")

ErrInvalidDefinitions is returned by LoadDefinitions when the file exists but has invalid JSON or colliding agent names after normalization. JSON errors are wrapped, so errors.As still recovers the parse position.

Functions

func Agents 

func Agents() []string

Agents lists every agent name this package knows: the recognized CLIs in [knownAgents] (most of which keep no transcript worth reading), the ones LoadDefinitions registered, and the ones RegisterSpec added. Sorted and deduplicated, so an agent several of those name appears once. Use Supported to tell which of them can be read.

func ConnectedTo 

func ConnectedTo(pid int, endpoints []netip.AddrPort) bool

ConnectedTo reports whether a process holds a connection to any of the given endpoints, which is how a monitor decides that an agent's tokens are already being counted somewhere else.

Example 
package main

import (
	"fmt"
	"net/netip"

	"github.com/maci0/toktop/agentusage"
)

func main() {
	engine, err := netip.ParseAddrPort("[::1]:11434")
	if err != nil {
		return
	}
	for _, p := range agentusage.Discover() {
		// A false answer means "cannot tell", which reads as "not connected".
		if agentusage.ConnectedTo(p.PID, []netip.AddrPort{engine}) {
			fmt.Printf("%s pid %d is talking to the engine\n", p.Tool, p.PID)
		}
	}
}

func DefinitionsPath 

func DefinitionsPath() string

DefinitionsPath is where agent definitions live by default. It follows gauntlet's location so one file serves both tools.

GAUNTLET_HOME is honored only when absolute, the same rule the XDG base directories get in openCodeDBPath and defaultKnownHostsPath. A relative value would place agents.json under whatever directory the run started in, where a missing file is not an error: the defined agents would simply never appear, looking like agents producing no tokens. Falling back to the documented default keeps the run reading the file it always read; the startup warning (warnIgnoredGauntletHome) names the ignored value.

func EnableOpenCodeDB 

func EnableOpenCodeDB(on bool) bool

EnableOpenCodeDB turns reading of opencode's SQLite session store on or off.

It is gated twice on purpose. The build tag `sqlite` decides whether the database driver is linked in at all, since it is a large dependency for one agent, and this switch decides whether a program that has it actually opens the operator's session database. Neither gate implies the other.

Call it before Watch or Supported for "opencode": a watcher built while the store is off cannot be turned on later. It reports whether this build can read it: false means the binary was compiled without `-tags sqlite`, and nothing was enabled.

func InputRate added in v0.8.0 

func InputRate(prev, cur Sample) (float64, bool)

InputRate returns billed prompt tokens per second between two samples, and whether it could be computed. Same rules as Rate: both samples need a timestamp, and no positive span or no growth means no rate, not a zero.

Example 
package main

import (
	"fmt"
	"time"

	"github.com/maci0/toktop/agentusage"
)

func main() {
	t0 := time.Unix(1_000_000, 0)
	prev := agentusage.Sample{Input: 80, At: t0}
	cur := agentusage.Sample{Input: 200, At: t0.Add(time.Second)}
	r, ok := agentusage.InputRate(prev, cur)
	fmt.Println(int(r), ok)
}

Output:
120 true

func LoadDefinitions 

func LoadDefinitions(path string) error

LoadDefinitions reads agent definitions from a JSON file, teaching this package about agents it was not compiled to know, including where they keep their transcripts: 

{"myagent": {"usage": {"roots": ["~/.myagent/sessions"]}}}

A missing file is not an error, since most machines have none. A malformed or unreadable one is: running with a half-loaded agent set is worse than refusing. The error names the file; errors.Is matches ErrInvalidDefinitions for invalid JSON or colliding agent names after normalization.

A definition may replace another definition, including one compiled into this build: the pi family is defined rather than adapted, so a file naming feynman with different roots redirects it. What it cannot displace is a registered adapter, built-in or installed by RegisterSpec, so a usage entry for an agent already read that way (claude, codex, dsh, ...) is skipped rather than registered, the same as one naming no roots. Registering it would leave SpecFor reporting roots no watcher reads. Use RegisterSpec to read such an agent elsewhere; it displaces the built-in for as long as it is held.

Names are canonicalized before registration, so two spellings that NFC reduces to one key (NFD "café" beside precomposed "café") would silently overwrite each other in defs. That overlap is refused instead: the file is ambiguous about which spec the surviving key should hold, and silently keeping whichever entry iterates last makes the loaded agent set depend on map order. The registry is left untouched. That one cause is also ErrCollidingDefinitions, so a caller can name it without reading the message.

Loading is additive per name, so a second load of the same agent replaces that agent rather than adding a second copy. ResetDefinitions drops everything this call added.

Example 
package main

import (
	"fmt"

	"github.com/maci0/toktop/agentusage"
)

func main() {
	err := agentusage.LoadDefinitions("/no/such/agents.json")
	fmt.Println(err)
}

Output:
<nil>

func MatchingEndpoints added in v0.6.0 

func MatchingEndpoints(pids []int, endpoints []netip.AddrPort) map[int]netip.AddrPort

MatchingEndpoints maps each pid to the first endpoint it holds a connection to. On Linux one pass over the kernel tables covers every process, so a dashboard watching N agents does not reread /proc/net/tcp N times (or N×M times when matching M engines one by one); where the platform has no shared-table reader, this falls back to one Peers call per pid.

Endpoints are matched on port plus address, with loopback spellings treated as equal: an engine advertised as 127.0.0.1:11434 and a connection to ::1:11434 are the same engine. The returned value is the advertised endpoint, not the peer's local spelling.

A pid with no connection to any of the endpoints is absent from the map, so look one up with the comma-ok form; the zero netip.AddrPort a bare index yields is not a result. As in Peers, an empty or unreadable result means "cannot tell", which a caller should read as "not connected".

Example

The engine-overlap check: endpoints an engine is advertised on, and the agents connected to any of them. A pid with no match is absent from the map. 

package main

import (
	"fmt"
	"net/netip"

	"github.com/maci0/toktop/agentusage"
)

func main() {
	engine, err := netip.ParseAddrPort("127.0.0.1:11434")
	if err != nil {
		return
	}
	for _, p := range agentusage.Discover() {
		// The map is keyed by pid, and holds the advertised endpoint a
		// process is connected to, not the peer's own spelling of it.
		for _, ep := range agentusage.MatchingEndpoints([]int{p.PID}, []netip.AddrPort{engine}) {
			fmt.Printf("%s pid %d already feeds %s\n", p.Tool, p.PID, ep)
		}
	}
}

func Peers 

func Peers(pid int) []netip.AddrPort

Peers lists the TCP endpoints a process is connected to.

An empty result means "cannot tell", which a caller should treat as "assume it is not the same engine" rather than as a statement about the process. On Linux this walks the process's open descriptors for socket inodes and looks each up in the kernel's TCP tables, all through procfs.

Example 
package main

import (
	"fmt"

	"github.com/maci0/toktop/agentusage"
)

func main() {
	for _, p := range agentusage.Discover() {
		for _, ep := range agentusage.Peers(p.PID) {
			fmt.Printf("%s pid %d is connected to %s\n", p.Tool, p.PID, ep)
		}
	}
}

func Rate 

func Rate(prev, cur Sample) (float64, bool)

Rate returns output tokens per second between two samples, and whether it could be computed at all. Both samples need a timestamp: a missing one is not a reading, and treating it as the zero instant would invent a rate off a first sample whose counter has already grown. It never extrapolates: without two readings and a positive span there is no rate to report. Prompt growth is InputRate.

Example 
package main

import (
	"fmt"
	"time"

	"github.com/maci0/toktop/agentusage"
)

func main() {
	t0 := time.Unix(1_000_000, 0)
	prev := agentusage.Sample{Output: 100, At: t0}
	cur := agentusage.Sample{Output: 350, At: t0.Add(time.Second)}
	r, ok := agentusage.Rate(prev, cur)
	fmt.Println(int(r), ok)
}

Output:
250 true

func RegisterSpec 

func RegisterSpec(tool string, spec Spec) error

RegisterSpec adds a transcript adapter for a defined agent. It returns ErrEmptyTool or an error wrapping ErrNoRoots when the spec cannot be used.

Example 
package main

import (
	"fmt"

	"github.com/maci0/toktop/agentusage"
)

func main() {
	err := agentusage.RegisterSpec("", agentusage.Spec{})
	fmt.Println(err)
}

Output:
usage spec needs an agent name
Example (FakeAgent)

The pattern a consumer's own tests need: register an agent this package does not ship, point it at a transcript the test writes, and take the registration back out so the next test in the binary does not inherit it. 

package main

import (
	"fmt"
	"os"
	"path/filepath"
	"time"

	"github.com/maci0/toktop/agentusage"
)

func main() {
	dir, err := os.MkdirTemp("", "agentusage-example")
	if err != nil {
		fmt.Println(err)
		return
	}
	defer os.RemoveAll(dir)
	if err := agentusage.RegisterSpec("fakeagent", agentusage.Spec{Roots: []string{dir}}); err != nil {
		fmt.Println(err)
		return
	}
	defer agentusage.UnregisterSpec("fakeagent")

	w := agentusage.Watch("fakeagent", "/home/me/project", time.Now())
	if w.Err() != nil {
		fmt.Println(w.Err())
		return
	}
	// A transcript already on disk when the watcher attached belongs to an
	// earlier run, so the record has to be written after it. The generic
	// reader takes counters under any of the names the supported agents use.
	record := []byte(`{"usage":{"input_tokens":30,"output_tokens":12}}` + "\n")
	if err := os.WriteFile(filepath.Join(dir, "session.jsonl"), record, 0o644); err != nil {
		fmt.Println(err)
		return
	}
	s := w.Poll()
	fmt.Println(s.Input, s.Output, s.Total)
}

Output:
30 12 42

func ResetDefinitions added in v0.15.0 

func ResetDefinitions()

ResetDefinitions drops every definition LoadDefinitions added, leaving the ones compiled into this build. It is the undo that call otherwise has none of, and the counterpart to UnregisterSpec: like it, the definitions registry is process-wide, so a program (or a test in one) that teaches this package an agent has to be able to take it back out.

Agents registered with RegisterSpec are adapters rather than definitions and are left alone; UnregisterSpec removes those.

Example

A test that loads a definitions file leaves the process-wide registry changed for every later test in the same binary, so it undoes the load. 

package main

import (
	"fmt"
	"os"
	"path/filepath"

	"github.com/maci0/toktop/agentusage"
)

func main() {
	dir, err := os.MkdirTemp("", "agentusage-example")
	if err != nil {
		fmt.Println(err)
		return
	}
	defer os.RemoveAll(dir)
	path := filepath.Join(dir, "agents.json")
	if err := os.WriteFile(path,
		[]byte(`{"myagent": {"usage": {"roots": ["~/.myagent/sessions"]}}}`), 0o644); err != nil {
		fmt.Println(err)
		return
	}
	if err := agentusage.LoadDefinitions(path); err != nil {
		fmt.Println(err)
		return
	}
	_, loaded := agentusage.SpecFor("myagent")
	agentusage.ResetDefinitions()
	_, after := agentusage.SpecFor("myagent")
	fmt.Println(loaded, after)
}

Output:
true false

func SetLogger added in v0.16.0 

func SetLogger(l *slog.Logger)

SetLogger sends the lines this package audits to l. The default is the process logger from slog.Default; a program that reads agents alongside its own output passes the logger it already writes to, so a failed walk is one line in the stream the program keeps rather than a second stream it does not. A nil logger restores the default.

The lines carry no program name, because which program is reading agents is the host's to say. A host that needs its own name in the message prepends it in the handler it passes.

func Supported 

func Supported(tool string) bool

Supported reports whether live usage can be read for an agent.

Example 
package main

import (
	"fmt"

	"github.com/maci0/toktop/agentusage"
)

func main() {
	for _, tool := range agentusage.Agents() {
		if agentusage.Supported(tool) {
			fmt.Println(tool)
		}
	}
}

func ThinkingRate added in v0.16.0 

func ThinkingRate(prev, cur Sample) (float64, bool)

ThinkingRate returns reasoning tokens per second between two samples, and whether it could be computed. Same rules as Rate, over the reasoning share of Output rather than all of it. An agent that does not report reasoning separately never grows it, so this reports no rate for one.

Example

Reasoning is the share of Output an agent reports separately, and it rates the same way, so a caller showing a thinking rate has one call to make. 

package main

import (
	"fmt"
	"time"

	"github.com/maci0/toktop/agentusage"
)

func main() {
	t0 := time.Unix(1_000_000, 0)
	prev := agentusage.Sample{Thinking: 20, At: t0}
	cur := agentusage.Sample{Thinking: 70, At: t0.Add(time.Second)}
	r, ok := agentusage.ThinkingRate(prev, cur)
	fmt.Println(int(r), ok)
}

Output:
50 true

func UnregisterSpec added in v0.15.0 

func UnregisterSpec(tool string) bool

UnregisterSpec removes the adapter RegisterSpec installed for an agent and restores the one it replaced, reporting whether a registration was there to remove. It is how a program that teaches this package an agent, or fakes one in its own tests, takes that back: the registry is process-wide, so without it every later test in the same binary inherits the spec.

The agent name is canonicalized like everywhere else, so " pi " and "pi" unregister the same agent. A name that was never registered removes nothing and reports false.

The restored adapter is a built-in one if a built-in was displaced, and the agent's loaded definition if there was no registered adapter to displace. A non-file source (opencode) is registered by EnableOpenCodeDB, not here, and is left alone. Watchers already running keep the adapter they attached with.

Example 
package main

import (
	"fmt"

	"github.com/maci0/toktop/agentusage"
)

func main() {
	err := agentusage.RegisterSpec("myagent", agentusage.Spec{
		Roots: []string{"~/.myagent/sessions"},
	})
	fmt.Println(err, agentusage.Supported("myagent"))
	agentusage.UnregisterSpec("myagent")
	fmt.Println(agentusage.Supported("myagent"))
}

Output:
<nil> true
false

Types

type Delta added in v0.15.0 

type Delta struct {
	// Output is generated tokens since the previous sample.
	Output int
	// Thinking is the reasoning share of Output, under the same rules.
	Thinking int
	// Input is billed prompt tokens since the previous sample.
	Input int
	// At is when the current sample was read, whether or not anything grew, so
	// a caller can stamp the interval the samples span.
	At time.Time
}

Delta is the growth between two consecutive samples: one interval's usage, where a Sample is the total observed since the watcher attached. A caller that reports events reports these, never the totals, or it bills the same tokens once per poll.

type Process 

type Process struct {
	// PID is the OS process identifier.
	PID int
	// Tool is the agent name (claude, codex, pi, …).
	Tool string
	// Dir is the process's working directory, which is what attributes a
	// transcript to it.
	Dir string
	// Started is when the process began, as far as the OS reports it; the
	// zero time when the platform does not report it.
	Started time.Time
}

Process is one running agent CLI.

func Discover 

func Discover() []Process

Discover lists the agent CLIs running on this machine.

On Linux this is a /proc walk: comm, cmdline, and cwd per process, plus starttime for matches; no subprocess. Every other user's processes simply fail the permission check. A nil result is not an error: /proc unreadable and nothing matched both return nil, and callers should surface either as no local agents.

func (Process) Watch added in v0.8.0 

func (p Process) Watch(since time.Time) *Watcher

Watch starts reading usage for this process, using its agent name and working directory. Equivalent to Watch(p.Tool, p.Dir, since).

Example 
package main

import (
	"fmt"
	"time"

	"github.com/maci0/toktop/agentusage"
)

func main() {
	for _, p := range agentusage.Discover() {
		if w := p.Watch(time.Now()); w.Err() == nil {
			fmt.Println(w.Tool(), w.Dir())
		}
	}
}

type Sample 

type Sample struct {
	// Output is generated tokens observed since attach.
	Output int
	// Thinking is the reasoning share of Output, when the agent reports it
	// separately: what the model spent before it wrote anything the user sees.
	Thinking int
	// Total is the largest per-request context size seen, not a sum: summing
	// those would count the same conversation once per turn. It is a level
	// rather than a running total, so it has no [Delta] field: a caller showing
	// a context window reads cur.Total off the current sample, and does not
	// report it as growth. A sample that only grows Total still reaches a Run
	// callback, and cur.Delta(prev) reports no growth for it, since no tokens
	// were spent.
	Total int
	// Input is billed prompt tokens, accrued per request the same way Output is.
	Input int
	// At is when the counters last changed, which is when the reading was
	// taken. A poll that observed nothing does not move it, so a stalled
	// agent's rate is averaged over the whole pause rather than over the
	// poll interval.
	At time.Time
}

Sample is cumulative usage observed since the watcher attached.

func (Sample) Delta added in v0.15.0 

func (s Sample) Delta(prev Sample) (Delta, bool)

Delta returns what grew from prev to the current sample, and whether anything did.

Counters only rise between two readings of the same watcher, so a sample smaller than the one before it is a transcript rewritten under the watcher: the figures it replaced are ones it no longer records, and counting them as growth bills the same tokens twice. That case reports no growth, which is what taking the current sample as the new baseline comes to. A caller passing the previous sample and keeping the current one has the whole re-baselining rule: 

if d, ok := cur.Delta(prev); ok {
	report(d)
}
prev = cur
Example 
package main

import (
	"fmt"
	"time"

	"github.com/maci0/toktop/agentusage"
)

func main() {
	t0 := time.Unix(1_000_000, 0)
	prev := agentusage.Sample{Output: 100, Input: 80, At: t0}
	cur := agentusage.Sample{Output: 350, Input: 200, At: t0.Add(time.Second)}

	d, ok := cur.Delta(prev)
	fmt.Println(ok, d.Output, d.Input)
	if r, ok := agentusage.Rate(prev, cur); ok {
		fmt.Println(int(r))
	}
}

Output:
true 250 120
250

func (Sample) Empty 

func (s Sample) Empty() bool

Empty reports whether nothing has been observed yet. Thinking-only samples count: an agent that reports reasoning without a billed output is still a reading, and callers that skip Empty samples must not drop it.

Example 
package main

import (
	"fmt"

	"github.com/maci0/toktop/agentusage"
)

func main() {
	fmt.Println(agentusage.Sample{}.Empty())
	fmt.Println(agentusage.Sample{Thinking: 12}.Empty())
}

Output:
true
false

type Spec 

type Spec struct {
	// Roots are directories to search, with ~ expanded. Blank entries are
	// ignored, and a spec with none left is not usable.
	//
	// {dir} in a root stands for the agent process's working directory, which
	// is what an agent keeping its transcripts inside the project it works in
	// needs (clanker keeps one under state/):
	//
	//	{"roots": ["{dir}/state"]}
	//
	// Roots name directories; the suffix chooses the files inside them.
	Roots []string `json:"roots"`
	// Suffix filters transcript files (default ".jsonl"). Surrounding
	// whitespace is trimmed, and a value left blank by that falls back to the
	// default rather than matching nothing.
	Suffix string `json:"suffix,omitempty"`
	// Suffixes matches several extensions, for an agent that writes more than
	// one (compressed by default, plain when compression is off). It replaces
	// Suffix when set, and blank entries among them are ignored.
	Suffixes []string `json:"suffixes,omitempty"`
	// Cumulative says the counters already include everything before them, so
	// the first value seen becomes a baseline. Default is per message.
	Cumulative bool `json:"cumulative,omitempty"`

	// HeaderCwd says the working directory appears once in a session header
	// rather than on every record, so ownership is decided from the head of
	// the file. Without it, a transcript whose usage lines carry no cwd is
	// attributed by location alone.
	HeaderCwd bool `json:"header_cwd,omitempty"`
}

Spec describes where a defined agent keeps its transcripts, so live usage works for agents this package was not compiled to know about (pi and the CLIs built on it, in-house wrappers). The records are parsed generically: any JSONL whose objects carry recognizable token counters works, and one whose objects do not simply reports nothing.

func SpecFor added in v0.15.0 

func SpecFor(tool string) (Spec, bool)

SpecFor reports the transcript location registered for an agent, whether it was compiled in (the pi family) or loaded by LoadDefinitions, and whether there is one at all. It is how a program that wrote a definitions file finds out what that file registered, since LoadDefinitions says nothing about entries it skipped and a skipped entry is otherwise indistinguishable from one that was never in the file.

The name is canonicalized like everywhere else, so " pi " and "pi" name the same agent. Roots come back as written: ~ and {dir} are expanded per process when a watcher walks them, which is why a spec is the right thing to show a person and the resolved paths are not.

An agent this package was compiled to read (claude, codex, dsh, …) is described by a built-in adapter rather than a definition, so SpecFor reports false for it; Supported is the question that covers every agent. That holds for one a definitions file tried to define as well, since LoadDefinitions skips an entry an adapter outranks.

Example 
package main

import (
	"fmt"

	"github.com/maci0/toktop/agentusage"
)

func main() {
	// A definitions file says nothing about the entries it skipped, so a
	// program that wrote one asks the registry what actually landed. Roots come
	// back as written, and the name is canonicalized like every other lookup.
	spec, ok := agentusage.SpecFor(" pi ")
	fmt.Println(ok, spec.Roots)
}

Output:
true [~/.pi/agent/sessions]

type Watcher 

type Watcher struct {
	// contains filtered or unexported fields
}

Watcher tails one agent's transcripts from the moment it attached.

func Watch 

func Watch(tool, dir string, since time.Time) *Watcher

Watch starts reading usage for one agent working in one directory.

tool is the agent name (claude, codex, crush, …). dir is the working directory that attributes transcripts to this process. since bounds database-backed agents (opencode, crush): only usage recorded after that instant is counted. File transcripts are always tailed from their attach-time end, so since does not rewind them; pass time.Now() at attach.

It returns nil when that agent keeps no readable transcript, which callers should treat as "no rate available" rather than an error. Err names that case for a caller that reports it: 

if w := agentusage.Watch(tool, dir, time.Now()); w.Err() != nil {
	// no readable usage for this agent
}

func (*Watcher) Dir 

func (w *Watcher) Dir() string

Dir is the working directory this watcher attributes usage to.

func (*Watcher) Err added in v0.15.0 

func (w *Watcher) Err() error

Err reports whether this watcher can read anything, and is the reason to show a caller who asked for one it cannot have.

A watcher is either usable or nil: Watch never returns one that is bound to nothing, so Err is nil on every live watcher and matches ErrUnsupportedTool on the nil one. Like Tool and Dir it is safe to call on the result without a nil check: 

if w := agentusage.Watch(tool, dir, time.Now()); w.Err() != nil {
	return w.Err() // the agent is known, but keeps nothing readable here
}

A nil watcher means the agent is unknown here, or keeps transcripts no build of this package can read, or has a definition naming no roots. The error says only that: the caller already knows which agent it asked about.

Example 
package main

import (
	"errors"
	"fmt"
	"time"

	"github.com/maci0/toktop/agentusage"
)

func main() {
	w := agentusage.Watch("nosuchagent", "", time.Now())
	fmt.Println(w == nil, errors.Is(w.Err(), agentusage.ErrUnsupportedTool))
}

Output:
true true

func (*Watcher) Poll 

func (w *Watcher) Poll() Sample

Poll reads whatever the transcripts have gained since the last read and returns the total. Callers use it for a final synchronous read once the agent has exited, since the last records land after the process is gone.

func (*Watcher) Run 

func (w *Watcher) Run(ctx context.Context, every time.Duration, onChange func(Sample))

Run polls until the context is canceled, calling onChange whenever the observed usage changes, growth or the drop a rewritten transcript causes. It is meant to run in its own goroutine.

onChange receives a Sample, which is a running total since the watcher attached, not the usage of the interval since the previous call. A caller that emits usage events therefore reports cur.Delta(prev) and keeps cur as the next baseline; printing the totals themselves bills the same tokens once per poll. Sample.Delta is that rule in one call, and the no-growth case is the re-baselining half: a sample that reports no growth is the one to measure from next time, whether the agent was quiet or a rewritten transcript replaced the figures.

every is how often the transcripts are re-read; a non-positive value uses DefaultPollInterval, which is also the value to pass for that default.

Example

Run reports the running total; Delta is what changed since the last report, so a caller that emits events never bills the same tokens twice. A delta that reports no growth is the sample to measure from next time, whatever the reason: a quiet agent, or a transcript rewritten under the watcher. 

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/maci0/toktop/agentusage"
)

func main() {
	w := agentusage.Watch("claude", "/home/me/project", time.Now())
	if w.Err() != nil {
		return // the agent keeps nothing this package can read
	}
	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
	defer cancel()
	var prev agentusage.Sample
	w.Run(ctx, agentusage.DefaultPollInterval, func(cur agentusage.Sample) {
		if d, ok := cur.Delta(prev); ok {
			fmt.Printf("%d output, %d prompt at %s\n", d.Output, d.Input, d.At.Format(time.TimeOnly))
		}
		prev = cur
	})
	// One final read after the agent has exited: the last records of a session
	// land once the process is gone, so Run's last callback is not the end of
	// the story.
	if d, ok := w.Poll().Delta(prev); ok {
		fmt.Printf("%d output, %d prompt at the end\n", d.Output, d.Input)
	}
}

func (*Watcher) Sample 

func (w *Watcher) Sample() Sample

Sample returns the usage observed so far.

func (*Watcher) SetNow added in v0.15.0 

func (w *Watcher) SetNow(fn func() time.Time)

SetNow overrides the clock that stamps published samples, and with them the event ids a caller derives from them. Call before Run: a simulated or frozen clock replays the same readings onto the same timeline, so a replayed run produces the same ids and the dashboard's id window drops the duplicates instead of counting them twice.

The windows this watcher applies to itself, the recency window and the transcript rescan interval, follow this clock too, so a run steps them instead of waiting them out. What it does not move is the other side of every comparison: file mtimes and the session-store timestamps that since bounds stay wall time, because that is the clock the filesystem and the stores record in. Anchor the injected clock to the wall instant the run started and the two agree.

Example 
package main

import (
	"time"

	"github.com/maci0/toktop/agentusage"
)

func main() {
	w := agentusage.Watch("claude", "/home/me/project", time.Now())
	// A frozen clock stamps every published sample with one instant, so a
	// replayed run derives the same event ids from the same readings.
	base := time.Date(2026, time.March, 1, 9, 0, 0, 0, time.UTC)
	w.SetNow(func() time.Time { return base })
}

func (*Watcher) Tool 

func (w *Watcher) Tool() string

Tool is the agent this watcher follows.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.