Something went wrong. Try again.
small gleam coding and persistent agent daemon with a detachable cli and a canvas ui forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322package daemon
import ( "encoding/json" "errors" "strings"
"albedo/cli/internal/daemon/protocol")
type wireChatEvent struct { CallID string StreamEvent}type sessionEventEnvelope struct { Type string `json:"type"` Sequence *int64 `json:"sequence"` Data json.RawMessage `json:"data"`}
func decodeChatEvent(raw json.RawMessage) (*wireChatEvent, error) { var envelope sessionEventEnvelope if err := decodeRequired(raw, &envelope, "type", "data"); err != nil { return nil, err } return decodeChatEnvelope(envelope, len(raw), nil)}
func decodeChatEnvelope(envelope sessionEventEnvelope, wireBytes int, content *eventContentReader) (*wireChatEvent, error) { if envelope.Type == "" || envelope.Data == nil || string(envelope.Data) == "null" { return nil, fieldError("event") } if envelope.Type != "reset" && envelope.Type != "failure" && (envelope.Sequence == nil || *envelope.Sequence < 1) { return nil, fieldError("event sequence") } event := StreamEvent{Type: EventType(envelope.Type)} switch envelope.Type { case "reset": var data struct { Reason string `json:"reason"` } if err := decodeRequired(envelope.Data, &data, "reason"); err != nil { return nil, err } switch data.Reason { case "initial", "generation_changed", "replay_unavailable", "recovery": event.Reason = data.Reason default: return nil, fieldError("reset reason") } case "status": var data protocol.SessionStatus if err := decodeRequired(envelope.Data, &data); err != nil { return nil, err } if err := validateSessionStatus(data); err != nil { return nil, err } status := statusValue(data, protocol.Kernel{}) event.Status = &status case "input": var data struct { Input protocol.Input `json:"input"` } if err := decodeRequired(envelope.Data, &data, "input"); err != nil { return nil, err } if err := validInput(data.Input); err != nil { return nil, err } receipt := InputReceipt(data.Input) event.Receipt = &receipt event.OperationID = data.Input.ID if data.Input.Turn != nil { event.Type = EventTurnMembership event.TurnID = data.Input.Turn.ID event.SubmissionIDs = []string{data.Input.ID} } case "text", "thinking": var data struct { RunID string `json:"run_id"` MessageID string `json:"message_id"` ElapsedMs *int64 `json:"elapsed_ms"` Text string `json:"text"` } if err := decodeRequired(envelope.Data, &data, "run_id", "message_id", "text"); err != nil { return nil, err } if data.RunID == "" || data.MessageID == "" { return nil, fieldError("message identity") } event.Text, event.TurnID, event.MessageID = data.Text, data.RunID, data.MessageID if data.ElapsedMs != nil { event.ElapsedMs = *data.ElapsedMs event.ElapsedObserved = true } case "message": var data struct { Entry protocol.HistoryEntry `json:"entry"` ReplacesLiveIDs []string `json:"replaces_live_ids"` } if err := decodeRequired(envelope.Data, &data, "entry"); err != nil { return nil, err } decoded, err := decodeHistoryEntries([]protocol.HistoryEntry{data.Entry}) if err != nil { return nil, err } if content != nil { decoded, err = hydrateHistory(content.ctx, content.conn, content.session, decoded) if err != nil { return nil, classifyStreamFailure(err) } } entries, err := historyEntryEvents(decoded[0]) if err != nil { return nil, err } if len(entries) == 0 { return nil, nil } event = entries[0] event.Replayed = false event.ReplacesLiveIDs = data.ReplacesLiveIDs case "note": var data struct { Text string `json:"text"` EntryID string `json:"entry_id"` Origin string `json:"origin"` Mail *protocol.MailMetadata `json:"mail"` } if err := decodeRequired(envelope.Data, &data, "entry_id", "origin", "text", "mail"); err != nil { return nil, err } event.Text, event.Source = data.Text, data.Origin case "tool_progress": if wireBytes > 8192 { return nil, errors.New("tool_progress exceeds 8 KiB") } var data struct { Progress json.RawMessage `json:"progress"` } if err := decodeRequired(envelope.Data, &data, "progress"); err != nil { return nil, err } progress, err := decodeToolProgress(data.Progress) if err != nil { return nil, err } event.Progress = progress case "tool": var data toolEventData if err := decodeRequired(envelope.Data, &data, "tool_call_id", "progress_call_id", "name", "arguments", "result", "trace", "content_complete", "reference"); err != nil { return nil, err } if data.ToolCallID == "" || data.ProgressCallID == "" || data.Name == "" { return nil, fieldError("tool identity") } if content != nil { if err := content.hydrateTool(&data); err != nil { return nil, classifyStreamFailure(err) } } event.ToolName, event.ProgressCallID = data.Name, data.ProgressCallID event.ToolStatus = data.Status event.ToolTitle = value(data.Title) arguments, _ := dynamicValue(data.Arguments) event.ToolArgs, _ = arguments.(map[string]any) if json.Unmarshal(data.Result, &event.ToolResult) != nil { event.ToolResult = string(data.Result) } event.ToolTrace = ParseToolTrace(data.Trace) event.Sequence = value(envelope.Sequence) return &wireChatEvent{CallID: data.ToolCallID, StreamEvent: event}, nil case "usage": var data protocol.Usage if err := decodeRequired(envelope.Data, &data); err != nil { return nil, err } event.Usage = usageValue(data) case "kernel": var data struct { Kernel protocol.Kernel `json:"kernel"` } if err := decodeRequired(envelope.Data, &data, "kernel"); err != nil { return nil, err } event.Kernel = &data.Kernel case "glances": var data struct { Glances []protocol.Glance `json:"glances"` } if err := decodeRequired(envelope.Data, &data, "glances"); err != nil { return nil, err } if len(data.Glances) > 16 { return nil, fieldError("glances") } event.Glances = glancesValue(data.Glances) case "committed": var data struct { HighWater int64 `json:"high_water"` ReplacesLiveIDs []string `json:"replaces_live_ids"` ReplacesAllLive bool `json:"replaces_all_live"` } if err := decodeRequired(envelope.Data, &data, "high_water", "replaces_live_ids", "replaces_all_live"); err != nil { return nil, err } if data.HighWater < 0 { return nil, fieldError("history high water") } if data.ReplacesLiveIDs == nil || len(data.ReplacesLiveIDs) > 256 { return nil, fieldError("live replacement identities") } for _, id := range data.ReplacesLiveIDs { if id == "" { return nil, fieldError("live replacement identity") } } event.Seq, event.ReplacesLiveIDs = data.HighWater, data.ReplacesLiveIDs event.ReplacesAllLive = data.ReplacesAllLive case "turn_completed": var data struct { RunID string `json:"run_id"` State string `json:"state"` InputIDs []string `json:"input_ids"` InputCount int64 `json:"input_count"` Truncated bool `json:"truncated"` } if err := decodeRequired(envelope.Data, &data, "run_id", "state", "input_ids", "input_count", "truncated"); err != nil { return nil, err } if data.RunID == "" || data.InputIDs == nil { return nil, fieldError("turn completion") } event.TurnID, event.SubmissionIDs = data.RunID, data.InputIDs if data.State == "interrupted" { event.Source = "interrupted" } else if data.State == "failed" || data.State == "abandoned" { event.Source = "failed" } else if data.State != "completed" { return nil, fieldError("turn state") } case "retry": var data struct { RunID string `json:"run_id"` Reason protocol.SafeReason `json:"reason"` } if err := decodeRequired(envelope.Data, &data, "run_id", "attempt", "reason", "delay_ms"); err != nil { return nil, err } event.TurnID, event.Text = data.RunID, data.Reason.Detail case "compacted": var data protocol.CompactionObservation if err := json.Unmarshal(envelope.Data, &data); err != nil { return nil, err } event.Evicted = int(data.EvictedEntries) event.Summary = data.Summary event.Strategy = value(data.Strategy) case "error": var data struct { RunID *string `json:"run_id"` Code string `json:"code"` Message string `json:"message"` } if err := decodeRequired(envelope.Data, &data, "run_id", "code", "message"); err != nil { return nil, err } if data.Code == "history_publication_failed" { return nil, &APIError{StatusCode: 503, Code: data.Code, Message: data.Message} } event.Text, event.TurnID, event.Reason = data.Message, value(data.RunID), data.Code case "invalidate": var data struct { Kind string `json:"kind"` URL string `json:"url"` } if err := decodeRequired(envelope.Data, &data, "kind", "url"); err != nil { return nil, err } var resources ResourceInvalidation switch data.Kind { case "session": resources.Session = true case "settings": resources.Settings = true case "catalog": resources.Catalog = true case "context": resources.Context = true case "extension": resources.Extension = true default: return nil, fieldError("invalidation kind") } if !strings.HasPrefix(data.URL, "/") || strings.HasPrefix(data.URL, "//") { return nil, fieldError("invalidation URL") } event.Invalidation = &resources case "failure": return nil, errors.New("failure must be a terminal batch") default: return nil, nil } event.Sequence = value(envelope.Sequence) return &wireChatEvent{StreamEvent: event}, nil}
type toolEventData struct { Status string `json:"status"` Reference *protocol.ContentReference `json:"reference"` ToolCallID string `json:"tool_call_id"` ProgressCallID string `json:"progress_call_id"` Name string `json:"name"` Title *string `json:"title,omitempty"` Arguments json.RawMessage `json:"arguments"` Result json.RawMessage `json:"result"` Trace json.RawMessage `json:"trace"` ContentComplete bool `json:"content_complete"`}