Documentation
¶
Overview ¶
Package attention implements the Attention Service: a server-side component that consumes the event bus of every active moa session and produces a priority-ordered "attention queue" of items already written for the ear.
It is the brain of the voice orchestrator. A dumb realtime client (browser today, native app tomorrow) connects to a single WebSocket and reads spoken briefings instead of the raw per-session bus; it never decides what matters.
Scope of this file set: P0 items — the things that STOP an agent and require the user: ask_user, permission_request, and error. No progress (P1/P2/P3), no digest, no LLM summaries, no coalescing. See tmp/plans/attention-service-design.md.
Concurrency model: a single goroutine (Service.loop) owns ALL mutable state (snapshots, queue, item states, generations, the active client). Nothing else mutates it. External reads (init, get_status) and bus events arrive as messages on channels; the loop replies with copies, never live pointers.
Index ¶
- Constants
- type AttentionItem
- type Briefing
- type ClientMsg
- type ClientSink
- type Config
- type ItemState
- type Kind
- type Priority
- type RiskLevel
- type RunTermination
- type ServerMsg
- type Service
- func (s *Service) Ack(itemID string)
- func (s *Service) AckForClient(c ClientSink, itemID string) bool
- func (s *Service) AckTerminationForClient(c ClientSink, terminationID string) bool
- func (s *Service) Attach(b bus.EventBus, sessionID, alias, title string, initialBrief ...SessionBrief) func()
- func (s *Service) ClearActiveClient(c ClientSink)
- func (s *Service) Close()
- func (s *Service) Roster() []SessionBrief
- func (s *Service) SetActiveClient(c ClientSink)
- func (s *Service) Snapshot() ServerMsg
- func (s *Service) SnapshotForClient(c ClientSink) (ServerMsg, bool)
- func (s *Service) Start()
- func (s *Service) Status() []AttentionItem
- func (s *Service) UpdateBrief(sessionID, attempting, progress string, updated time.Time)
- func (s *Service) UpdateMeta(sessionID, alias, title string)
- type SessionActivity
- type SessionBrief
- type TerminationRef
Constants ¶
const ProtocolVersion = 1
ProtocolVersion is sent in init so a client can refuse a mismatched server.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type AttentionItem ¶
type AttentionItem struct {
ID string `json:"id"` // stable attention-item id (att_%d)
Priority Priority `json:"priority"` // P0..P3
Kind Kind `json:"kind"` // ask | permission | error | ...
SessionID string `json:"session_id"` // origin session
Alias string `json:"alias"` // pronounceable session name for speech
Spoken string `json:"spoken"` // text written for the ear
State ItemState `json:"state"`
CreatedAt time.Time `json:"created_at"`
// RefID is moa's real pending id for P0 ask/permission (perm_%d / ask_%d),
// empty for errors. The client resolves via existing /ask & /permission
// endpoints; resolution flows back through the bus, not through this service.
RefID string `json:"ref_id,omitempty"`
// Risk metadata — only meaningful for KindPermission. Populated by the
// deterministic risk parser (assessRisk). Never softened by any model.
RiskLevel RiskLevel `json:"risk_level,omitempty"`
RiskFlags []string `json:"risk_flags,omitempty"`
// Verbatim is the exact command/args, ALWAYS captured even when not read
// aloud, so a client can offer "read it to me literally".
Verbatim string `json:"verbatim,omitempty"`
}
AttentionItem is one thing worth telling the user about. The Spoken field is already written for the ear. RefID carries moa's real pending ID (perm_%d / ask_%d) so the client can resolve it through the EXISTING HTTP endpoints — the Attention Service itself resolves nothing.
type Briefing ¶
type Briefing struct {
Priority Priority `json:"priority"` // P1 (terminal) or P2 (progress)
Kind Kind `json:"kind"` // run_ok | goal_ended | goal_stalled | verify_fail
SessionID string `json:"session_id"` // origin session
Alias string `json:"alias"` // pronounceable session name
Spoken string `json:"spoken"` // text written for the ear
// Termination is set only for a successful run completion. It lets a voice
// client ask for the full result through the existing messages endpoint.
Termination *RunTermination `json:"termination,omitempty"`
}
Briefing is an ephemeral spoken note about progress (a run finished, a goal ended, a verify failed). Unlike an AttentionItem it carries no id, no state, no ref — the chief-of-staff mentions it and moves on.
type ClientMsg ¶
type ClientMsg struct {
Type string `json:"type"` // "ack" | "ack_termination" | "get_status"
RequestID string `json:"request_id"` // echoed on errors
ItemID string `json:"item_id"` // for ack
TerminationID string `json:"termination_id"` // for ack_termination
}
ClientMsg is one message received from the active voice client. ack confirms an attention item; ack_termination confirms a run notice after it was spoken; get_status requests an authoritative init snapshot.
type ClientSink ¶
type ClientSink interface {
// Send delivers one server->client message. Returns false if the sink is
// dead (the loop then forgets it).
Send(msg ServerMsg) bool
// ID identifies the connection so the loop can detect supersession.
ID() uint64
// Close revokes this sink. A superseded guardian must not retain a live
// control channel.
Close()
}
ClientSink is the minimal interface the loop uses to push messages to the one active voice client. The WS handler implements it. Sends must not block the loop: implementations buffer and drop the CONNECTION (not the message) on overflow — init repairs state on reconnect.
type Config ¶
type Config struct {
// Lang is the resolved briefing language code (e.g. "en", "es"); from
// core.GetSTTLanguage. Empty -> English.
Lang string
// OnUndelivered is an optional future push hook. It is called when a P0
// item is born without an active guardian; callers can use it to wake a
// paired phone without putting APNs policy in this package.
OnUndelivered func(AttentionItem)
}
Config holds the Service dependencies. All fields optional except as noted.
type ItemState ¶
type ItemState string
ItemState is the lifecycle state of an attention item. These are distinct on purpose (see design §2.2): being read aloud (announced) is not the same as the client confirming it spoke it (acked), which is not the same as the underlying request being answered (resolved).
const ( // StatePending: created, not yet delivered to a client. StatePending ItemState = "pending" // StateAnnounced: delivered to the active client to be spoken. StateAnnounced ItemState = "announced" // StateAcked: the client confirmed it relayed the item to the user. Stops // escalation/retry but does NOT resolve the underlying request. StateAcked ItemState = "acked" // StateResolved: the underlying permission/ask was answered (learned from // the bus), or the error cleared. Terminal. StateResolved ItemState = "resolved" )
type Kind ¶
type Kind string
Kind is the semantic type of an attention item. Used for dedup signatures and client rendering.
const ( KindAsk Kind = "ask" // agent is asking the user a question KindPermission Kind = "permission" // agent requests permission to run a tool KindError Kind = "error" // session entered the error state // Progress/terminal kinds. These are EPHEMERAL briefings, not // tracked P0 items: the chief-of-staff telling you how things are going. KindRunOK Kind = "run_ok" // a run finished successfully (P2) KindGoalEnded Kind = "goal_ended" // a goal loop stopped (P1) KindGoalStalled Kind = "goal_stalled" // a goal made no progress (P1) KindVerifyFail Kind = "verify_fail" // auto-verify reported failures (P1) )
type Priority ¶
type Priority int
Priority ranks how urgently an item needs the user's attention. Only emits P0; higher-numbered levels are defined for later phases so the wire protocol is stable.
const ( // P0Blocking: the agent is STOPPED waiting for the user (ask_user, // permission_request, error). Interrupt now. P0Blocking Priority = 0 // P1Terminal: a run finished with an error / a goal ended. P1Terminal Priority = 1 // P2Progress: a run finished OK / a goal iteration was satisfied. P2Progress Priority = 2 // P3Ambient: subagents, tasks, cost. P3Ambient Priority = 3 )
type RiskLevel ¶
type RiskLevel string
RiskLevel is the conservative danger classification of a permission's command, computed by a DETERMINISTIC parser (never by an LLM). Over-classifies on doubt: a command we can't confidently parse is treated as at least medium.
type RunTermination ¶
type RunTermination struct {
ID string `json:"id"`
SessionID string `json:"session_id"`
Alias string `json:"alias"`
Spoken string `json:"spoken"`
Summary string `json:"summary"`
CreatedAt time.Time `json:"created_at"`
Ref TerminationRef `json:"ref"`
}
RunTermination is the durable, at-least-once completion notice for one successful agent run. It is not an AttentionItem: it is neither actionable nor resolvable. Ref identifies the transcript containing the full answer; GET /api/sessions/{session_id}/messages is the existing detail endpoint.
type ServerMsg ¶
type ServerMsg struct {
Type string `json:"type"` // "init" | "attention" | "item_update" | "briefing" | "roster" | "inactive" | "error"
V int `json:"v,omitempty"`
// init
Items []AttentionItem `json:"items,omitempty"` // all unresolved P0 items
Sessions []SessionBrief `json:"sessions,omitempty"` // the roster of attached sessions (init + roster)
// attention (a new item) / item_update (state change of an existing item)
Item *AttentionItem `json:"item,omitempty"`
// briefing: an ephemeral progress/terminal note. It is NOT
// tracked or resolvable. The sole exception is a successful run completion,
// whose Termination metadata is retained for recovery in init.
Briefing *Briefing `json:"briefing,omitempty"`
// terminations contains successful run completions awaiting explicit client
// acknowledgement. It appears only in init and is replaced by the next init,
// like Items and Sessions. Clients must deduplicate by ID and send
// ack_termination only after speaking the notice.
Terminations []RunTermination `json:"terminations,omitempty"`
// error
RequestID string `json:"request_id,omitempty"`
Code string `json:"code,omitempty"`
Message string `json:"message,omitempty"`
}
ServerMsg is one message sent to the active voice client.
type Service ¶
type Service struct {
// contains filtered or unexported fields
}
Service is the attention brain. Construct with New, wire sessions with Attach, tear down with Close.
func (*Service) Ack ¶
Ack marks an item acknowledged (the client relayed it). Stops escalation; does not resolve. Idempotent.
func (*Service) AckForClient ¶
func (s *Service) AckForClient(c ClientSink, itemID string) bool
AckForClient acknowledges an item only if c is still the active guardian. It prevents a superseded socket from issuing commands after replacement.
func (*Service) AckTerminationForClient ¶
func (s *Service) AckTerminationForClient(c ClientSink, terminationID string) bool
AckTerminationForClient confirms a termination only if c is still active.
func (*Service) Attach ¶
func (s *Service) Attach(b bus.EventBus, sessionID, alias, title string, initialBrief ...SessionBrief) func()
Attach registers a session and returns an unsubscribe-style detach func plus the bus handler wiring. The caller is responsible for subscribing the returned handler to the session bus and calling the returned func on session teardown. Generation tokens make late events harmless after detach.
aliasTitle is (alias, title); seedInfo is the current pending approval state (may be zero). Returns a detach func to append to the session's unsub list.
func (*Service) ClearActiveClient ¶
func (s *Service) ClearActiveClient(c ClientSink)
ClearActiveClient removes c as the sink if it is still the active one (on WS disconnect). No-op if a newer client already superseded it.
func (*Service) Close ¶
func (s *Service) Close()
Close stops the loop and releases resources. Idempotent. After Close the loop emits nothing further. Safe to call from Manager.Shutdown.
func (*Service) Roster ¶
func (s *Service) Roster() []SessionBrief
Roster returns the current attached-session roster. It is public so HTTP clients can consume the same global attention state as the guardian.
func (*Service) SetActiveClient ¶
func (s *Service) SetActiveClient(c ClientSink)
SetActiveClient makes c the single active voice sink and immediately sends it an authoritative init. Any previous client is told it's inactive.
func (*Service) Snapshot ¶
Snapshot returns the authoritative guardian init payload. get_status uses this same shape so the caller can replace, rather than merge, local state.
func (*Service) SnapshotForClient ¶
func (s *Service) SnapshotForClient(c ClientSink) (ServerMsg, bool)
SnapshotForClient returns the authoritative init payload only while c is the active guardian. This prevents a superseded socket from serving control responses after it has been replaced.
func (*Service) Status ¶
func (s *Service) Status() []AttentionItem
Status returns the current unresolved items (for get_status).
func (*Service) UpdateBrief ¶
UpdateBrief updates a session's LLM-generated status prose. No-op if the session is not attached. Refreshes the client roster; actionable state is still maintained independently from live bus events.
func (*Service) UpdateMeta ¶
UpdateMeta updates a session's spoken alias and human title (e.g. after auto-title generation or a manual rename). No-op if the session isn't attached. Refreshes the client roster.
type SessionActivity ¶
type SessionActivity struct {
Kind string `json:"kind"` // "subagent" | "tool"
Detail string `json:"detail,omitempty"` // subagent task, or tool target (bash command, etc.), bounded
Tool string `json:"tool,omitempty"` // tool name when kind=="tool" (bash, edit, fetch_content, ...)
Model string `json:"model,omitempty"` // child model when kind=="subagent"
Count int `json:"count,omitempty"` // number of active subagents when kind=="subagent"
}
SessionActivity is the freshest live action a session is performing, as structured data the voice client narrates (never prose the server localizes).
type SessionBrief ¶
type SessionBrief struct {
SessionID string `json:"session_id"`
Alias string `json:"alias"`
Title string `json:"title"`
State string `json:"state"` // idle | running | permission | error
PendingAsks int `json:"pending_asks"` // count of unanswered questions
PendingPerm int `json:"pending_perms"` // count of unresolved permissions
Activity *SessionActivity `json:"activity,omitempty"`
Attempting string `json:"brief_attempting,omitempty"`
Progress string `json:"brief_progress,omitempty"`
BriefUpdated time.Time `json:"brief_updated,omitzero"`
}
SessionBrief is the voice client's view of one live agent session: enough to name it, know its state, and address orders to it (via the existing HTTP endpoints). It carries moa's real session id so the client can POST to /api/sessions/{id}/... The client never derives what an agent is doing from the bus — the server hands it this compact, authoritative view.
type TerminationRef ¶
type TerminationRef struct {
SessionID string `json:"session_id"`
RunGen uint64 `json:"run_gen"`
MessagesURL string `json:"messages_url"`
}
TerminationRef points a client at the completed run's transcript. RunGen distinguishes successive runs in a session; the messages endpoint exposes the complete owner-authorized conversation rather than a special voice API.