acp

package
v0.0.21 Latest Latest
Warning

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

Go to latest
Published: Oct 18, 2026 License: MIT Imports: 15 Imported by: 0

Documentation

Overview

Package acp provides an ACP (Agent Client Protocol) client for connecting to ACP-compatible agents and receiving live session update streams.

The Client connects to a running agent process via stdio JSON-RPC, parses session update messages, and forwards them to a SessionHandler for indexing, storage, or forwarding.

Design notes:

  • ACP wire protocol: newline-delimited JSON-RPC 2.0 over stdio
  • Session state is tracked per sessionId with goroutine-safe access
  • No leaked goroutines: context cancellation stops the read loop cleanly
  • All ACP types are defined here; acp-go-sdk is not a dependency (we define our own aligned types to avoid the external SDK requirement)

Package acp — IndexingSessionHandler implements SessionHandler by flushing each ACP session update to the audit trail immediately (per-update mode). Hook events are fired at session start and end via the injected hooks.Manager.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func RegisteredFormats

func RegisteredFormats() []string

RegisteredFormats returns a sorted slice of all compile-time registered format names.

RegisteredFormats is safe for concurrent use.

Types

type Adapter

type Adapter interface {
	// Parse converts a raw record into a SessionUpdate.
	//
	// Returns a non-nil error if the record is malformed, missing required
	// fields, or cannot be decoded. The error must identify:
	//   (1) what field or byte caused the failure,
	//   (2) why decoding failed (unexpected type, missing key, etc.),
	//   (3) how to fix it (expected format or value).
	Parse(record []byte) (SessionUpdate, error)

	// Format returns the canonical name of the format this adapter handles.
	// Example values: "claude-jsonl", "opencode-json".
	Format() string
}

Adapter converts agent-specific transcript record bytes into a SessionUpdate.

Each Adapter handles one agent format (e.g. "claude-jsonl", "opencode-json"). Format() returns the canonical format name, which is used as the registration key and is also the string callers pass to GetAdapter.

Parse receives a single raw record (one line for JSONL formats, one JSON object for structured formats). It must return a descriptive error that includes the byte offset or field name of any parsing failure so that callers can locate the problem in the source file.

Implementations must be safe for concurrent use; no shared mutable state.

func GetAdapter

func GetAdapter(format string) (Adapter, error)

GetAdapter retrieves the registered adapter for the given format name.

Returns an error if no adapter has been registered for format. The error lists all currently registered format names to help the caller choose a valid option.

GetAdapter is safe for concurrent use; the underlying map is read-only.

func NewClaudeAdapter

func NewClaudeAdapter() Adapter

NewClaudeAdapter returns a Claude JSONL adapter instance.

func NewOpenCodeAdapter

func NewOpenCodeAdapter() Adapter

NewOpenCodeAdapter returns an OpenCode JSON adapter instance.

type Client

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

Client connects to ACP-compatible agents and receives live session updates.

Call Connect to start an agent process and begin receiving updates. The client is goroutine-safe; SessionStats and SessionCount may be called concurrently with Connect.

Lifecycle:

client := NewClient(handler)
err := client.Connect(ctx, "claude", "--mcp-server", "...")
// Connect blocks until ctx is cancelled or the agent exits
err = client.Disconnect()

func NewClient

func NewClient(handler SessionHandler) *Client

NewClient creates an ACP client that forwards session updates to handler. handler must not be nil.

func (*Client) Connect

func (c *Client) Connect(ctx context.Context, agentCmd string, agentArgs ...string) error

Connect starts the agent process identified by agentCmd and agentArgs, reads newline-delimited JSON-RPC messages from its stdout, and forwards parsed SessionUpdate events to the configured SessionHandler.

Connect blocks until:

  • ctx is cancelled (clean shutdown)
  • the agent process exits
  • a handler returns an error

After Connect returns, call Disconnect to clean up the process. Calling Connect again on the same Client after it returns is not supported; create a new Client instead.

func (*Client) Disconnect

func (c *Client) Disconnect() error

Disconnect signals the agent process to stop and waits for cleanup. It is safe to call Disconnect after Connect returns; it is a no-op in that case.

func (*Client) SessionCount

func (c *Client) SessionCount() int

SessionCount returns the total number of sessions seen (including ended ones). Safe for concurrent use.

func (*Client) SessionStats

func (c *Client) SessionStats(sessionId string) (*SessionStats, error)

SessionStats returns a snapshot of the stats for the given sessionId. Returns an error if the session has not been observed.

type ContentBlock

type ContentBlock struct {
	// Type identifies the block type (e.g. "text", "tool_use", "tool_result", "thinking").
	Type string `json:"type"`
	// Text holds the text payload for "text" and "thinking" blocks, matching the
	// ACP spec wire key "text".
	Text string `json:"text,omitempty"`
	// ID is the tool call identifier for "tool_use" and "tool_result" blocks.
	ID string `json:"id,omitempty"`
	// Name is the tool name for "tool_use" blocks.
	Name string `json:"name,omitempty"`
	// Input holds tool input as raw JSON for "tool_use" blocks.
	Input json.RawMessage `json:"input,omitempty"`
	// RawContent holds tool result content for "tool_result" blocks in wire format.
	RawContent json.RawMessage `json:"rawContent,omitempty"`
}

ContentBlock represents a single piece of content in a session update. Aligned with the ACP content block model, whose text blocks are {"type":"text","text":"..."} — i.e. the spec wire key is "text".

Text is the single canonical field, named for the ACP/Anthropic spec wire key "text", so it decodes and round-trips with the standard library: no custom UnmarshalJSON/MarshalJSON. This replaced an earlier dual Content/Text pair that left callers (e.g. the indexer) guessing which field held the payload.

type IndexingSessionHandler

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

IndexingSessionHandler implements SessionHandler by indexing and persisting each update immediately on HandleUpdate (per-update flush mode).

All methods are safe for concurrent invocation (multiple sessions may run in parallel within a single durable step).

Lifecycle per session:

  1. First HandleUpdate call → fire HookSessionStarted, index update, persist to trail.
  2. Subsequent HandleUpdate calls → index update, persist to trail (no extra hooks).
  3. HandleSessionEnd → fire HookSessionEnded → clean up session tracking state.

func NewIndexingSessionHandler

func NewIndexingSessionHandler(
	indexer *SharedIndexer,
	trail audit.Trail,
	hooksMgr *hooks.Manager,
	epochId string,
) *IndexingSessionHandler

NewIndexingSessionHandler constructs a ready-to-use IndexingSessionHandler.

  • indexer must not be nil.
  • trail must not be nil.
  • hooksMgr may be nil (hooks are optional; nil manager is a no-op).
  • epochId is embedded in every hook payload so consumers can correlate events.

func (*IndexingSessionHandler) EntriesRecorded

func (h *IndexingSessionHandler) EntriesRecorded() int

EntriesRecorded returns the total number of SessionEntry rows persisted by this handler across all sessions so far. Safe for concurrent use.

func (*IndexingSessionHandler) HandleSessionEnd

func (h *IndexingSessionHandler) HandleSessionEnd(ctx context.Context, sessionId string, reason StopReason) error

HandleSessionEnd finalises the session identified by sessionId.

Steps (in order):

  1. Dispatch HookSessionEnded (best-effort).
  2. Record last session ID and stop reason.
  3. Remove the session from the tracking map.

No bulk indexing occurs here — entries are already persisted by HandleUpdate. Returns nil unless the context is cancelled.

func (*IndexingSessionHandler) HandleUpdate

func (h *IndexingSessionHandler) HandleUpdate(ctx context.Context, update SessionUpdate) error

HandleUpdate indexes the update immediately and persists it to the audit trail.

On the first update for a session, HookSessionStarted is dispatched via the Manager (if non-nil). Hook errors are treated as best-effort: logged via slog.Warn but do not cause HandleUpdate to return an error.

Returns an error only if the context is already cancelled, indexing fails, or the trail write fails.

func (*IndexingSessionHandler) LastSessionId

func (h *IndexingSessionHandler) LastSessionId() string

LastSessionId returns the session ID from the most recent HandleSessionEnd call. Returns an empty string if no session has ended yet. Safe for concurrent use.

func (*IndexingSessionHandler) LastStopReason

func (h *IndexingSessionHandler) LastStopReason() StopReason

LastStopReason returns the StopReason from the most recent HandleSessionEnd call. Returns an empty StopReason if no session has ended yet. Safe for concurrent use.

type SessionHandler

type SessionHandler interface {
	// HandleUpdate processes a single session update.
	// Called for every message received from the agent, in order.
	// Returning an error causes the client to stop processing.
	HandleUpdate(ctx context.Context, update SessionUpdate) error

	// HandleSessionEnd is called when a session ends (stop_reason present).
	// sessionId is the session that ended. reason is the stop reason.
	HandleSessionEnd(ctx context.Context, sessionId string, reason StopReason) error
}

SessionHandler processes live ACP session updates.

Implementations handle indexing (SharedIndexer), storage, event forwarding, or hook dispatch. All methods must be safe for concurrent invocation.

type SessionStats

type SessionStats struct {
	SessionId   string
	UpdateCount int
	StartTime   time.Time
	LastUpdate  time.Time
	Ended       bool
	EndReason   StopReason // zero value if session is still active
	EndTime     time.Time  // zero value if session is still active
}

SessionStats is the public view of a session's state.

type SessionUpdate

type SessionUpdate struct {
	// SessionId uniquely identifies the agent session.
	SessionId string `json:"sessionId"`
	// Role is the message author: "user", "assistant", or "tool".
	Role string `json:"role"`
	// Content holds the ordered content blocks for this update.
	Content []ContentBlock `json:"content,omitempty"`
	// ToolCalls holds extracted tool invocations from this update.
	// Populated by Adapter.Parse; the live ACP client derives these from Content.
	ToolCalls []ToolCall `json:"toolCalls,omitempty"`
	// StopReason is present when the session ends; empty during normal updates.
	StopReason StopReason `json:"stopReason,omitempty"`
	// Usage holds token usage for this update (may be nil for intermediate updates).
	Usage *UsageStats `json:"usage,omitempty"`
	// TokensIn is the number of input tokens consumed (convenience accessor).
	// Adapters should populate this directly; live client reads from Usage.
	TokensIn int `json:"tokensIn,omitempty"`
	// TokensOut is the number of output tokens produced (convenience accessor).
	TokensOut int `json:"tokensOut,omitempty"`
	// Timestamp is the wall-clock time the update was emitted, in milliseconds since epoch.
	Timestamp int64 `json:"timestamp,omitempty"`
	// IsError signals that this update represents an error turn.
	IsError bool `json:"isError,omitempty"`
	// EntryId is the provider-native message ID.
	EntryId string `json:"entryId,omitempty"`
	// ParentEntryId links tool_result turns back to their tool_use origin.
	ParentEntryId string `json:"parentEntryId,omitempty"`
}

SessionUpdate represents a single streamed update from an ACP agent. Each update corresponds to one message turn in the session.

For the live ACP client (Connect), updates come from JSON-RPC params. For adapter-parsed updates (Claude JSONL, OpenCode JSON), fields are populated by the respective Adapter.Parse implementation.

type SharedIndexer

type SharedIndexer struct{}

SharedIndexer converts a slice of ACP SessionUpdate events into a flat list of protocol.SessionEntry rows suitable for persistence in the audit trail.

The indexer is stateless and safe for concurrent use. Each call to Index processes only the provided updates; no state is retained between calls.

Conversion rules:

  • Each SessionUpdate with no ToolCalls produces one depth=0 SessionEntry.
  • Each ToolCall within a SessionUpdate produces one depth=1 SessionEntry whose ParentIndex refers to the depth=0 entry generated for the same update.
  • ContentPreview is truncated to 500 characters.
  • ToolNamesCsv collects all unique tool names for the update (depth=0 entry).

func NewSharedIndexer

func NewSharedIndexer() *SharedIndexer

NewSharedIndexer constructs a ready-to-use SharedIndexer.

func (*SharedIndexer) Index

func (idx *SharedIndexer) Index(updates []SessionUpdate) ([]protocol.SessionEntry, error)

Index converts the provided SessionUpdate slice into protocol.SessionEntry rows.

Returns a non-nil error only if an individual update cannot be converted (malformed data). Partial results up to the first error are not returned; the caller receives either the full result set or an error.

Index guarantees: len(entries) >= len(updates) (each update contributes at least one depth=0 entry plus one depth=1 entry per ToolCall).

type StopReason

type StopReason string

StopReason classifies why an ACP session ended. Values align with the ACP protocol stop_reason field.

const (
	// StopReasonEndTurn means the agent completed its turn normally.
	StopReasonEndTurn StopReason = "end_turn"
	// StopReasonMaxTokens means the session hit a token limit.
	StopReasonMaxTokens StopReason = "max_tokens"
	// StopReasonToolUse means the agent is paused at a tool call.
	StopReasonToolUse StopReason = "tool_use"
	// StopReasonError means the session ended due to an error.
	StopReasonError StopReason = "error"
	// StopReasonCancelled means the session was cancelled by the client.
	StopReasonCancelled StopReason = "cancelled"
)

type ToolCall

type ToolCall struct {
	// ToolCallId is the correlation ID linking tool_use to tool_result.
	ToolCallId string `json:"toolCallId"`
	// ToolKind classifies the tool category (function, bash, file, etc.).
	ToolKind ToolKind `json:"toolKind"`
	// ToolName is the name of the tool or function invoked.
	ToolName string `json:"toolName"`
	// ToolInput is the JSON-encoded input arguments supplied to the tool.
	ToolInput string `json:"toolInput,omitempty"`
	// ToolOutput is the JSON-encoded result returned by the tool (if completed).
	ToolOutput string `json:"toolOutput,omitempty"`
}

ToolCall represents a tool invocation and its result within a SessionUpdate.

type ToolKind

type ToolKind string

ToolKind classifies the category of a tool call for indexing and display.

const (
	// ToolKindFunction represents a standard function/capability call.
	ToolKindFunction ToolKind = "function"
	// ToolKindBash represents a shell command execution.
	ToolKindBash ToolKind = "bash"
	// ToolKindFile represents a file system read or write operation.
	ToolKindFile ToolKind = "file"
	// ToolKindSearch represents a search or grep-style operation.
	ToolKindSearch ToolKind = "search"
	// ToolKindWeb represents a web fetch or search operation.
	ToolKindWeb ToolKind = "web"
	// ToolKindUnknown is the fallback for unrecognized tool categories.
	ToolKindUnknown ToolKind = "unknown"
)

type UsageStats

type UsageStats struct {
	InputTokens  int `json:"input_tokens"`
	OutputTokens int `json:"output_tokens"`
}

UsageStats holds token consumption for a session update.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL