Documentation
¶
Overview ¶
Package live is the view-agnostic transport for an attached Managed Agents session. It backfills history, follows the SSE stream without a gap, de-duplicates by event id, folds agent.message deltas into preview events, keeps everything in display order and reconnects with backoff. The TUI consumes a Conn and the local web viewer a Threads (a Conn per thread), both as Snapshot() plus a channel of Updates; nothing here knows about rendering.
An event is pending until the server has processed it. A pending event is either a queued user event (sent, not yet processed) or a streaming preview (an agent.message still receiving deltas).
conn.go holds the Conn, its consumer contract and the sources it follows, follow.go the loop that keeps it attached, order.go the display order, preview.go the streaming previews, send.go the writes, and threads.go the multi-thread fan-out.
Index ¶
- type Conn
- func (c *Conn) Close()
- func (c *Conn) Confirm(ctx context.Context, toolUseID string, allow bool, denyMessage string) error
- func (c *Conn) Interrupt(ctx context.Context) error
- func (c *Conn) SendMessage(ctx context.Context, text string) error
- func (c *Conn) Session() *anthropic.BetaManagedAgentsSession
- func (c *Conn) Snapshot() []Event
- func (c *Conn) Updates() <-chan Update
- type Event
- type Thread
- type ThreadID
- type ThreadLog
- type ThreadUpdate
- type Threads
- func (t *Threads) Close()
- func (t *Threads) Confirm(ctx context.Context, toolUseID string, allow bool, denyMessage string) error
- func (t *Threads) Events(id ThreadID) []Event
- func (t *Threads) Interrupt(ctx context.Context) error
- func (t *Threads) SendMessage(ctx context.Context, text string) error
- func (t *Threads) Session() *anthropic.BetaManagedAgentsSession
- func (t *Threads) Snapshot() (threads []Thread, logs []ThreadLog)
- func (t *Threads) Updates() <-chan ThreadUpdate
- type Update
- type UpdateKind
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Conn ¶
type Conn struct {
// contains filtered or unexported fields
}
Conn follows one event log, the session's own or one thread's, and publishes it as Snapshot() plus Updates().
func Open ¶
Open loads the session synchronously so callers fail fast on a bad id or credentials; backfill and streaming then proceed in the background. The Conn lives until ctx is done or Close is called — pass the command's root context so Ctrl-C tears it down.
func (*Conn) Confirm ¶
Confirm answers a pending tool call. denyMessage is only sent with a denial. The verdict is stored as a queued placeholder before the POST so the call stops reading as awaiting approval at once; the server's echo (or a failure) then replaces it.
func (*Conn) Interrupt ¶
Interrupt posts a user.interrupt, asking the agent to stop its current turn.
func (*Conn) SendMessage ¶
SendMessage posts a user.message. SendMessage, Interrupt and Confirm block on the network and then on delivering an Update, so never call them from the goroutine that drains Updates().
func (*Conn) Session ¶
func (c *Conn) Session() *anthropic.BetaManagedAgentsSession
Session returns a copy of the session as of the newest event folded into it.
type Event ¶
type Event = anthropic.BetaManagedAgentsSessionEventUnion
Event is the List endpoint's union: every variant's fields flattened, with ProcessedAt.IsZero() marking a pending event.
type Thread ¶
type Thread = anthropic.BetaManagedAgentsSessionThread
Thread is one execution thread of a multi-agent session as threads.list reports it; the primary thread has an empty ParentThreadID.
type ThreadID ¶
type ThreadID = string
ThreadID names one event log of a multi-agent session: "" is the primary thread (the session's own stream), anything else a child thread's id. It is not an event's session_thread_id: events cross-posted to the session stream carry a child's id yet belong to thread "".
type ThreadUpdate ¶
type ThreadUpdate struct {
ThreadID ThreadID
Update
// KindThreads: the complete thread list, primary included.
Threads []Thread
}
ThreadUpdate is an Update tagged with the thread it applies to; on Reordered, rebuild that thread from Events(ThreadID). KindSession and KindThreads only ever come with ThreadID "".
type Threads ¶
type Threads struct {
// contains filtered or unexported fields
}
Threads follows a session and every thread it spawns: the primary Conn plus one child Conn per thread that threads.list reports, discovered whenever the primary stream shows thread activity. Consumers see one channel of ThreadUpdates and a Snapshot of every thread's log; writes go to the primary.
func OpenThreads ¶
OpenThreads fails fast like Open; thread discovery then runs in the background for as long as the primary Conn lives.
func (*Threads) Close ¶
func (t *Threads) Close()
Close stops every Conn and waits until Updates is closed.
func (*Threads) Confirm ¶
func (t *Threads) Confirm(ctx context.Context, toolUseID string, allow bool, denyMessage string) error
Confirm answers a pending tool call through the primary thread.
func (*Threads) Events ¶
Events returns one thread's events in display order: the rebuild source for a Reordered ThreadUpdate. nil for an id threads.list has not reported.
func (*Threads) SendMessage ¶
SendMessage posts a user.message to the primary thread.
func (*Threads) Session ¶
func (t *Threads) Session() *anthropic.BetaManagedAgentsSession
Session returns the primary Conn's session.
func (*Threads) Snapshot ¶
Snapshot returns the last thread list (empty until the session shows any thread activity) and every thread's events: primary first, then children by creation time. Children that have stopped keep their events.
func (*Threads) Updates ¶
func (t *Threads) Updates() <-chan ThreadUpdate
Updates closes once the primary Conn has stopped and every child with it.
type Update ¶
type Update struct {
Kind UpdateKind
// KindEvent: a whole event, never a raw stream frame. A pending
// agent.message is a streaming preview refined by later updates with the
// same id and finally replaced by the processed event. Empty when the
// update only signals Reordered.
Event Event
// Reordered: the ordered list changed in a way the contract cannot
// express (an insert ahead of a processed event or a streaming preview,
// a known id moving, or dropped previews).
Reordered bool
// KindSession: the refreshed session after a session.updated or status event.
Session *anthropic.BetaManagedAgentsSession
// KindConn: stream connectivity. Err says why it dropped (errors.As it to
// *anthropic.Error for API failures; 401/403/404 are fatal and are
// followed by Updates closing). Backfilled is set once, on the update that
// follows delivery of the full initial history.
Connected bool
Err error
Backfilled bool
}
Update is one change to what Snapshot()/Session() return, or to connectivity.
Consumer contract: every KindEvent is an upsert by Event.ID — replace the event you hold with that id; an id unknown to you goes last, except that queued user events always stay behind everything else. When Reordered is set (on any Kind), discard your list and rebuild from Snapshot(). A consumer that lags and rebuilds may briefly re-apply older updates; letting a pending event never overwrite a processed one avoids the only visible artefact, and the list always converges at quiescence.
type UpdateKind ¶
type UpdateKind int
UpdateKind says which fields of an Update carry the change.
const ( // KindEvent: Update.Event was inserted or replaced. KindEvent UpdateKind = iota // KindSession: Update.Session is the refreshed session. KindSession // KindConn: stream connectivity changed. KindConn // KindThreads comes only from Threads: the thread list in ThreadUpdate.Threads changed. KindThreads )
Source Files