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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176package daemon
import ( "context" "encoding/json" "io" "net/http"
"albedo/cli/internal/daemon/protocol")
func NewACPInput(session, token string, blocks []json.RawMessage) (*OperationHandle, error) { id, err := operationID() if err != nil { return nil, err } payload, err := json.Marshal(protocol.ACPPromptInputRequest{Kind: "acp_prompt", Blocks: blocks}) if err != nil { return nil, err } handle := &OperationHandle{id: id, kind: "acp_prompt", sessionID: session} handle.operation = operation{Name: "admit ACP input", Body: json.RawMessage(payload), Headers: http.Header{"Albedo-Attachment-Token": {token}}, Policy: receiptRecovery, Handle: handle, BuildRequest: func(base string, body io.Reader) (*http.Request, error) { return protocol.NewPutInputRequestWithBody(base, session, id, &protocol.PutInputParams{AlbedoAttachmentToken: &token}, "application/json", body) }} return handle, nil}
// CreateACPSession replays the exact creation request when its acknowledgement is lost.func CreateACPSession(ctx context.Context, conn *Connection, token string, request protocol.ACPSessionRequest) (string, error) { id, err := operationID() if err != nil { return "", err } payload, err := json.Marshal(request) if err != nil { return "", err } handle := &OperationHandle{id: id, kind: "creation", sessionID: id} op := operation{Name: "create ACP session", Body: json.RawMessage(payload), Handle: handle, Headers: http.Header{"Albedo-Attachment-Token": {token}, "If-None-Match": {"*"}}, Policy: receiptRecovery, BuildRequest: func(base string, body io.Reader) (*http.Request, error) { return protocol.NewCreateSessionRequestWithBody(base, id, &protocol.CreateSessionParams{IfNoneMatch: "*", AlbedoAttachmentToken: &token}, "application/json", body) }} handle.operation = op for range 2 { err = executeMutation(ctx, conn, op, []int{200, 201}, func(_ []byte, _ int) error { return nil }) if err == nil { return id, nil } if _, ok := err.(*UncertainOutcomeError); !ok { return "", err } } return "", err}func AttachACP(ctx context.Context, conn *Connection, id, token string, request protocol.ACPAttachRequest) (protocol.ACPAttachment, error) { var result protocol.ACPAttachment err := executeMutation(ctx, conn, operation{Name: "attach ACP", Body: request, Policy: noRecovery, BuildRequest: func(base string, body io.Reader) (*http.Request, error) { return protocol.NewAttachACPSessionRequestWithBody(base, id, &protocol.AttachACPSessionParams{AlbedoAttachmentToken: token}, "application/json", body) }}, []int{200}, func(data []byte, _ int) error { return decodeRequired(data, &result) }) return result, err}func RenewACP(ctx context.Context, conn *Connection, id, token string) error { return executeMutation(ctx, conn, operation{Name: "renew ACP", Body: struct{}{}, Policy: noRecovery, BuildRequest: func(base string, body io.Reader) (*http.Request, error) { return protocol.NewRenewACPSessionRequestWithBody(base, id, &protocol.RenewACPSessionParams{AlbedoAttachmentToken: token}, "application/json", body) }}, []int{204}, func(_ []byte, _ int) error { return nil })}func ReleaseACP(ctx context.Context, conn *Connection, id, token string) error { return executeMutation(ctx, conn, operation{Name: "release ACP", Policy: noRecovery, BuildRequest: func(base string, _ io.Reader) (*http.Request, error) { return protocol.NewReleaseACPSessionRequest(base, id, &protocol.ReleaseACPSessionParams{AlbedoAttachmentToken: token}) }}, []int{204}, func(_ []byte, _ int) error { return nil })}
func (c *ChatClient) SeedCursor(cursor protocol.Cursor) { c.mu.Lock() defer c.mu.Unlock() c.afterSeq = cursor.Sequence c.afterGeneration = cursor.Generation}func (c *ChatClient) CancelInput(ctx context.Context, id, token string) (protocol.InputCancellation, error) { var result protocol.InputCancellation params := &protocol.CancelInputParams{} if token != "" { params.AlbedoAttachmentToken = &token } err := executeMutation(ctx, c.conn, operation{Name: "cancel input", Body: struct{}{}, Policy: noRecovery, BuildRequest: func(base string, body io.Reader) (*http.Request, error) { return protocol.NewCancelInputRequestWithBody(base, c.agentID, id, params, "application/json", body) }}, []int{200}, func(data []byte, _ int) error { if err := decodeRequired(data, &result); err != nil { return err } if result.Input.ID != id { return fieldError("input identity") } switch result.Result { case "cancelled", "interrupt_requested", "shared_running", "not_pending": default: return fieldError("cancellation result") } return validInput(result.Input) }) return result, err}
// OutputBoundaries exposes byte counts for reset reconciliation, never content.func (s Session) OutputBoundaries() []protocol.ACPOutputBoundary { boundaries := make([]protocol.ACPOutputBoundary, 0, len(s.wire.ActiveOutput)) for _, output := range s.wire.ActiveOutput { boundaries = append(boundaries, protocol.ACPOutputBoundary{LiveID: output.MessageID, UTF8Bytes: output.Bytes}) } return boundaries}
// ListACPSessions reads one server-filtered page without acquiring attachments.func ListACPSessions(ctx context.Context, conn *Connection, cwd, cursor *string) (protocol.SessionPage, error) { params := protocol.ListSessionsParams{Acp: new(true), Workspace: cwd, Limit: new(int64(50)), Next: cursor} var page protocol.SessionPage err := executeRead(ctx, conn, operation{Name: "list ACP sessions", Policy: readRecovery, BuildRequest: func(base string, _ io.Reader) (*http.Request, error) { return protocol.NewListSessionsRequest(base, ¶ms) }}, func(data []byte) error { return decodeRequired(data, &page) }) return page, err}
// ACPHistoryCursor retains the entry offset encoded in the daemon's token.type ACPHistoryCursor struct { Token string Position int64 Done bool}
// ACPHistory reads one display page, excluding entries beyond the captured cutoff.func ACPHistory(ctx context.Context, conn *Connection, session string, cursor ACPHistoryCursor, cutoff int64) ([]StreamEvent, ACPHistoryCursor, error) { params := protocol.GetHistoryParams{Limit: new(int64(100))} if cursor.Token == "" { params.After = new(int64(0)) } else { params.Next = &cursor.Token } wire, err := readHistory(ctx, conn, session, params) if err != nil { return nil, cursor, err } rows := []protocol.HistoryEntry{} next := ACPHistoryCursor{Position: cursor.Position} if wire.Newer != nil { next.Token = *wire.Newer } seen := map[string]bool{} for _, row := range wire.Items { if row.Position <= 0 || row.Position < next.Position || seen[row.ID] { return nil, cursor, fieldError("ascending history position") } if row.Position > cutoff { next.Done = true break } seen[row.ID] = true rows = append(rows, row) next.Position = row.Position } if wire.Newer == nil { if wire.HighWater < cutoff { return nil, cursor, fieldError("history cutoff unavailable") } next.Done = true } entries, err := decodeHistoryEntries(rows) if err != nil { return nil, cursor, err } entries, err = hydrateHistory(ctx, conn, session, entries) if err != nil { return nil, cursor, err } events, err := historyEvents(entries) return events, next, err}