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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658package acp
import ( "bytes" "context" "crypto/sha256" "encoding/json" "errors" "fmt" "hash" "slices" "strings" "sync" "time" "unicode/utf8"
"albedo/cli/internal/app" "albedo/cli/internal/daemon" "albedo/cli/internal/daemon/protocol")
// Reconciliation keeps bounded identity metadata and output digests, not text.const maxReconciliationEntries = 4096const maxReconciliationIDBytes = 1024
type streamedPrefix struct { digest hash.Hash bytes int}
func appendPrefix(prefix *streamedPrefix, text string) *streamedPrefix { if prefix == nil { prefix = &streamedPrefix{digest: sha256.New()} } _, _ = prefix.digest.Write([]byte(text)) prefix.bytes += len(text) return prefix}
type pendingPrompt struct { handle *daemon.OperationHandle request request blocks []json.RawMessage admitted bool cancelling bool finishing bool cancelSent bool}type toolState uint8
const ( toolUnseen toolState = iota toolAnnounced toolCompleted)
// mu protects session state. Observation and replay handlers require the lock.// pauseObservation and resumeObservation acquire it.type session struct { bufferFailure error ctx context.Context failure error boundaries map[string]int64 // withheld names unfinished messages whose earlier chunks the editor missed. // Live chunks stay hidden until commit supplies the complete message. withheld map[string]bool client *daemon.ChatClient tools map[string]toolState replayCancel context.CancelFunc ready <-chan struct{} live map[string]*streamedPrefix adapter *adapter successor *pendingPrompt stop context.CancelFunc durable map[string]bool active *pendingPrompt models app.ACPModels id string turn string cursor protocol.Cursor inherited []string buffered []daemon.StreamEvent lease protocol.ACPLease historyCutoff int64 bufferedBytes int mu sync.Mutex uncertain bool buffering bool configuring bool}
func newSession(a *adapter, attachment protocol.ACPAttachment) *session { ctx, cancel := context.WithCancel(a.ctx) current := &session{ adapter: a, ctx: ctx, stop: cancel, id: attachment.SessionID, cursor: attachment.Cursor, client: a.backend.Observer(attachment.SessionID, attachment.Cursor), lease: attachment.Lease, historyCutoff: attachment.HistoryCutoff, inherited: slices.Clone(attachment.PendingInputIDs), live: map[string]*streamedPrefix{}, boundaries: map[string]int64{}, durable: map[string]bool{}, tools: map[string]toolState{}, withheld: map[string]bool{}, } if attachment.ActiveTurn != nil { current.turn = attachment.ActiveTurn.TurnID if attachment.ActiveTurn.AcpInputID != nil { current.inherited = append(current.inherited, *attachment.ActiveTurn.AcpInputID) } } for _, boundary := range attachment.OutputBoundaries { if len(boundary.LiveID) > maxReconciliationIDBytes { current.lose("observation_uncertain", "Output identity exceeds reconciliation limit") break } current.boundaries[boundary.LiveID] = boundary.UTF8Bytes current.limitReconciliation() if current.uncertain { break } } current.limitReconciliation() return current}func (s *session) start() { ready := make(chan struct{}) s.ready = ready var subscribed sync.Once go func() { err := s.client.StreamSubscribed(s.ctx, func(event daemon.StreamEvent) error { s.mu.Lock() defer s.mu.Unlock() return s.consumeLocked(event) }, func() { subscribed.Do(func() { close(ready) }) }) if s.ctx.Err() == nil { s.mu.Lock() s.lose("observation_uncertain", fmt.Sprintf("Session stream ended: %v", err)) s.mu.Unlock() } }()
go func() { interval := time.Duration(s.lease.HeartbeatMs) * time.Millisecond if interval <= 0 { interval = 5 * time.Second } lease := time.Duration(s.lease.LeaseMs) * time.Millisecond renewed := time.Now() timer := time.NewTicker(interval) defer timer.Stop() for { select { case <-s.ctx.Done(): return case <-timer.C: err := s.adapter.backend.Renew(s.ctx, s.id) if err == nil { renewed = time.Now() continue } // A transient failure is retried until the lease would have run // out; the daemon saying the attachment is gone ends it at once. api, refused := errors.AsType[*daemon.APIError](err) if (refused && api.Code == "attachment_lost") || time.Since(renewed) >= lease { s.mu.Lock() s.lose("attachment_lost", "The attachment was lost; resume the session") s.mu.Unlock() return } } } }()}func (s *session) prompt(message request, blocks []json.RawMessage) { s.mu.Lock() defer s.mu.Unlock() if s.buffering || s.configuring { s.adapter.respond(message.ID, nil, problem(-32602, "busy", "Session setup is in progress")) return } if s.uncertain { s.adapter.respond(message.ID, nil, problem(-32603, "observation_uncertain", "Resume the session before sending another prompt")) return } prompt := &pendingPrompt{request: message, blocks: blocks} if s.active != nil { // A finishing prompt has its response queued, so the editor may // already have it and send the next prompt before the response's // write barrier clears the active prompt: hold that one as the // successor instead of refusing it as busy. if (s.active.cancelling || s.active.finishing) && s.successor == nil { s.successor = prompt return } s.adapter.respond(message.ID, nil, problem(-32602, "busy", "A prompt is already active")) return } if s.turn != "" || len(s.inherited) > 0 { if s.successor == nil { s.successor = prompt } else { s.adapter.respond(message.ID, nil, problem(-32602, "busy", "A prompt is already waiting")) } return } s.admit(prompt)}func (s *session) admit(prompt *pendingPrompt) { handle, err := s.adapter.backend.Prepare(s.id, prompt.blocks) if err != nil { s.adapter.respond(prompt.request.ID, nil, err) return } prompt.handle = handle s.active = prompt go func() { select { case <-s.ready: case <-s.ctx.Done(): return } err := s.adapter.backend.Submit(s.ctx, handle) s.mu.Lock() defer s.mu.Unlock() if s.active != prompt || prompt.finishing { return } if err != nil { if _, uncertain := errors.AsType[*daemon.UncertainOutcomeError](err); uncertain && prompt.admitted { return } s.lose("observation_uncertain", err.Error()) return } prompt.admitted = true if prompt.cancelling { s.cancelInput(handle.ID(), prompt) } }()}func (s *session) cancelPrompt() { s.mu.Lock() defer s.mu.Unlock() if s.successor != nil { s.adapter.respond(s.successor.request.ID, map[string]any{"stopReason": "cancelled"}, nil) s.successor = nil } if s.active != nil { prompt := s.active prompt.cancelling = true if prompt.admitted && !prompt.finishing { s.cancelInput(prompt.handle.ID(), prompt) } return } for _, id := range s.inherited { s.cancelInput(id, nil) }}func (s *session) cancelInput(id string, prompt *pendingPrompt) { if prompt != nil { if prompt.cancelSent { return } prompt.cancelSent = true } go func() { receipt, err := s.adapter.backend.Cancel(s.ctx, s.id, id) s.mu.Lock() defer s.mu.Unlock() if prompt != nil && (s.active != prompt || prompt.finishing) { return } defer s.admitSuccessor() if err != nil { s.lose("observation_uncertain", "Cancellation could not be reconciled") return } switch receipt.Result { case "cancelled": s.inherited = removeID(s.inherited, id) if prompt != nil { s.finish("cancelled", nil) } case "interrupt_requested": case "not_pending": if receipt.Input.Delivery != nil && *receipt.Input.Delivery == "cancelled" { s.inherited = removeID(s.inherited, id) if prompt != nil { s.finish("cancelled", nil) } return } // A finished turn is finished by the stream: the receipt can run // ahead of the turn's last output, which must be sent first. case "shared_running": s.lose("observation_uncertain", "Daemon violated exclusive ACP input contract") default: s.lose("observation_uncertain", "Unknown cancellation result") } }()}func removeID(ids []string, id string) []string { return slices.DeleteFunc(ids, func(value string) bool { return value == id })}func (s *session) lose(reason, message string) { if s.active == nil && s.adapter.diagnostics != nil { fmt.Fprintln(s.adapter.diagnostics, message) } if s.buffering { s.failBufferingLocked(problem(-32603, reason, message)) } s.uncertain = true s.clearReconciliation() s.failure = problem(-32603, reason, message) if s.active != nil { s.finish("", s.failure) } if s.successor != nil { s.adapter.respond(s.successor.request.ID, nil, s.failure) s.successor = nil }}func (s *session) finish(reason string, failure error) { prompt := s.active if prompt == nil || prompt.finishing { return } if prompt.cancelling && failure != nil { reason, failure = "cancelled", nil } prompt.finishing = true barrier, err := s.adapter.writer.enqueue(s.adapter.response(prompt.request.ID, map[string]any{"stopReason": reason}, failure), true) if err != nil { s.adapter.cancel() return } go func() { select { case <-barrier: case <-s.ctx.Done(): return case <-s.adapter.writer.done: return } s.mu.Lock() defer s.mu.Unlock() if s.active != prompt { return } s.active = nil s.clearReconciliation() s.turn = "" s.failure = nil s.admitSuccessor() }()}func (s *session) admitSuccessor() { if s.successor == nil || s.active != nil || s.turn != "" || len(s.inherited) > 0 { return } next := s.successor s.successor = nil if s.uncertain { s.adapter.respond(next.request.ID, nil, problem(-32603, "observation_uncertain", "Resume required")) } else { s.admit(next) }}func (s *session) complete(state string) { switch state { case "interrupted": s.finish("cancelled", nil) case "failed", "abandoned": failure := s.failure if failure == nil { failure = problem(-32603, state, "The turn "+state) } s.finish("", failure) default: s.finish("end_turn", nil) }}func (s *session) observe(event daemon.StreamEvent) { if s.ctx.Err() != nil || s.uncertain || event.Replayed { return } for _, id := range []string{event.MessageID, event.EntryID, event.ProgressCallID} { if len(id) > maxReconciliationIDBytes { s.lose("observation_uncertain", "Output identity exceeds reconciliation limit") return } } defer s.limitReconciliation() if event.Type == daemon.EventReset { if s.reconcileReset(event.Snapshot) { return } s.lose("observation_uncertain", "The session stream reset; resume required") return } if event.Sequence > 0 { s.cursor.Sequence = event.Sequence } if event.SnapshotOrigin { return } if s.active != nil && s.active.finishing { return } if s.active != nil && event.OperationID == s.active.handle.ID() && event.Receipt != nil && event.Receipt.Admission == "accepted" { s.active.admitted = true if s.active.cancelling { s.cancelInput(s.active.handle.ID(), s.active) } } if event.Type == daemon.EventTurnMembership { if s.active != nil && slices.Contains(event.SubmissionIDs, s.active.handle.ID()) { s.turn = event.TurnID s.active.admitted = true if s.active.cancelling { s.cancelInput(s.active.handle.ID(), s.active) } } for _, id := range s.inherited { if slices.Contains(event.SubmissionIDs, id) { s.turn = event.TurnID } } } if s.turn == "" || event.TurnID != "" && event.TurnID != s.turn { return } switch event.Type { case daemon.EventText, daemon.EventThinking: if event.EntryID != "" { s.durableMessage(event) return } s.live[event.MessageID] = appendPrefix(s.live[event.MessageID], event.Text) if len(s.withheld) > 0 { // A later segment can belong to the same durable message as a captured // prefix. Hold it too until commit establishes the complete order. s.withheld[event.MessageID] = true return } kind := "agent_message_chunk" if event.Type == daemon.EventThinking { kind = "agent_thought_chunk" } s.chunk(kind, event.MessageID, event.Text) case daemon.EventMessage: s.durableMessage(event) case daemon.EventToolProgress: if event.Progress == nil { return } id := event.Progress.CallID switch s.tools[id] { case toolUnseen: s.announceTool(id, event.Progress.Name) case toolAnnounced: s.update(map[string]any{"sessionUpdate": "tool_call_update", "toolCallId": id, "status": "in_progress"}) case toolCompleted: // Late progress cannot reopen a completed tool. } case daemon.EventTool: if event.ProgressCallID == "" || event.ToolStatus != "completed" && event.ToolStatus != "failed" { s.lose("observation_uncertain", "Missing durable tool outcome") return } s.finishTool(event) case daemon.EventError: reason := event.Reason if reason == "" { reason = "turn_failed" } s.failure = problem(-32603, reason, event.Text) case daemon.EventRetry: s.failure = nil case daemon.EventCommitted: for _, id := range event.ReplacesLiveIDs { delete(s.live, id) delete(s.boundaries, id) delete(s.withheld, id) } if event.ReplacesAllLive { clear(s.live) clear(s.boundaries) clear(s.withheld) } case daemon.EventTurnCompleted: if event.TurnID != s.turn { return } s.inherited = slices.DeleteFunc(s.inherited, func(id string) bool { return slices.Contains(event.SubmissionIDs, id) }) if s.active != nil { s.complete(event.Source) } else { s.turn = "" s.failure = nil } s.clearReconciliation() s.admitSuccessor() }}func (s *session) chunk(kind, id, text string) { if text == "" { return } if err := s.adapter.writer.enqueueText(textChunk{SessionID: s.id, Kind: kind, MessageID: id, Text: text}); err != nil { s.adapter.cancel() }}
func (s *session) announceTool(id, title string) { if id == "" || len(id) > maxReconciliationIDBytes { s.lose("observation_uncertain", "Missing tool progress identity") return } if s.tools[id] != toolUnseen { return } s.tools[id] = toolAnnounced s.update(map[string]any{"sessionUpdate": "tool_call", "toolCallId": id, "title": title, "kind": "other", "status": "in_progress"})}
// The result may belong to both load replay and the following live cursor.func (s *session) finishTool(event daemon.StreamEvent) { if s.tools[event.ProgressCallID] == toolCompleted { return } s.announceTool(event.ProgressCallID, event.ToolName) if s.uncertain { return } s.tools[event.ProgressCallID] = toolCompleted s.update(toolDisplay(event))}
func (s *session) update(update map[string]any) { if _, err := s.adapter.writer.enqueue(map[string]any{"jsonrpc": "2.0", "method": "session/update", "params": map[string]any{"sessionId": s.id, "update": update}}, false); err != nil { s.adapter.cancel() }}
func (s *session) durableMessage(event daemon.StreamEvent) { id := event.MessageID if id == "" { id = event.EntryID } if id == "" { s.lose("observation_uncertain", "Missing durable message identity") return } if s.durable[id] { return } offset := 0 var unsent strings.Builder for _, liveID := range event.ReplacesLiveIDs { start := offset boundary, hasBoundary := s.boundaries[liveID] prefix, hasLive := s.live[liveID] if !hasBoundary && !hasLive { s.lose("observation_uncertain", "Durable replacement names unknown live output") return } if boundary < 0 || boundary > int64(len(event.Text)-offset) { s.lose("observation_uncertain", "Invalid output boundary") return } offset += int(boundary) if !utf8.ValidString(event.Text[offset:]) { s.lose("observation_uncertain", "Invalid output boundary") return } if hasLive { if prefix.bytes > len(event.Text)-offset { s.lose("observation_uncertain", "Durable text does not extend streamed output") return } sum := sha256.Sum256([]byte(event.Text[offset : offset+prefix.bytes])) if !bytes.Equal(sum[:], prefix.digest.Sum(nil)) { s.lose("observation_uncertain", "Durable text does not extend streamed output") return } offset += prefix.bytes }
if s.withheld[liveID] { unsent.WriteString(event.Text[start:offset]) } } s.durable[id] = true kind := "agent_message_chunk" if event.Type == daemon.EventThinking { kind = "agent_thought_chunk" } unsent.WriteString(event.Text[offset:]) s.chunk(kind, id, unsent.String()) for _, liveID := range event.ReplacesLiveIDs { delete(s.live, liveID) delete(s.boundaries, liveID) delete(s.withheld, liveID) }}
// A reset at the exact observed cursor can be checked without replaying any// snapshot content. A gap or generation change cannot prove duplicate-free output.func (s *session) reconcileReset(snapshot *daemon.Session) bool { if snapshot == nil || snapshot.Cursor == nil || *snapshot.Cursor != s.cursor { return false } if s.turn != "" && (snapshot.Status.RunID == nil || *snapshot.Status.RunID != s.turn) { return false } expected := map[string]int64{} for id, prefix := range s.live { expected[id] = int64(prefix.bytes) + s.boundaries[id] } for id, boundary := range s.boundaries { if _, exists := expected[id]; !exists { expected[id] = boundary } } outputs := snapshot.OutputBoundaries() if len(outputs) != len(expected) { return false } for _, output := range outputs { count, exists := expected[output.LiveID] if !exists || count != output.UTF8Bytes { return false } } return true}
func (s *session) clearReconciliation() { clear(s.live) clear(s.boundaries) clear(s.withheld) clear(s.tools) clear(s.durable)}
func (s *session) limitReconciliation() { if !s.uncertain && len(s.live)+len(s.boundaries)+len(s.withheld)+len(s.tools)+len(s.durable) > maxReconciliationEntries { s.lose("observation_uncertain", "Output reconciliation exceeds its identity limit; resume required") }}