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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302package daemon
import ( "context" "crypto/rand" "encoding/json" "errors" "fmt" "io" "net/http" "reflect" "time"
"albedo/cli/internal/daemon/protocol")
// OperationHandle freezes the chosen resource ID and exact admission payload.// Retries always use this same input or session resource.type OperationHandle struct { id string operation operation kind string sessionID string}
func (handle *OperationHandle) ID() string { return handle.id }func (handle *OperationHandle) IsCreation() bool { return handle.kind == "creation" }
func operationID() (string, error) { var id [16]byte if _, err := rand.Read(id[:]); err != nil { return "", fmt.Errorf("create resource ID: %w", err) } stamp := uint64(time.Now().UnixMilli()) for i := 5; i >= 0; i-- { id[i] = byte(stamp) stamp >>= 8 } id[6] = id[6]&0x0f | 0x70 id[8] = id[8]&0x3f | 0x80 return fmt.Sprintf("%x-%x-%x-%x-%x", id[:4], id[4:6], id[6:8], id[8:10], id[10:]), nil}
func NewSubmission(session string, fields SubmissionRequest) (*OperationHandle, error) { id := fields.SubmissionID if id == "" { var err error id, err = operationID() if err != nil { return nil, err } } kind := fields.Type if kind == "" || kind == "user" { kind = "message" } var body any switch kind { case "message": if fields.Content == nil { return nil, errors.New("message input requires text") } uploaded := make([]protocol.ImageUpload, len(fields.Images)) for i, image := range fields.Images { uploaded[i] = protocol.ImageUpload{MimeType: string(image.MimeType), Data: image.Data} } body = protocol.MessageInputRequest{Kind: kind, Text: *fields.Content, Images: uploaded, Pastes: fields.Pastes, ClientID: optionalText(fields.ClientID)} case "continue": body = protocol.ContinueInputRequest{Kind: kind, ClientID: optionalText(fields.ClientID)} case "skill": body = protocol.SkillInputRequest{Kind: kind, CandidateID: fields.Name, CatalogRevision: fields.CatalogRevision, Arguments: fields.Arguments, ClientID: optionalText(fields.ClientID)} case "command": body = protocol.CommandInputRequest{Kind: kind, CommandID: fields.Name, Arguments: fields.CommandArguments, ClientID: optionalText(fields.ClientID)} default: return nil, fmt.Errorf("unknown input kind %q", kind) } payload, err := json.Marshal(body) if err != nil { return nil, err } handle := &OperationHandle{id: id, kind: kind, sessionID: session} handle.operation = operation{Name: "admit input", BuildRequest: func(base string, body io.Reader) (*http.Request, error) { return protocol.NewPutInputRequestWithBody(base, session, id, "application/json", body) }, Body: json.RawMessage(payload), Policy: receiptRecovery, Handle: handle} return handle, nil}
func NewCreation(fields CreateSessionRequest) (*OperationHandle, error) { kind := fields.Kind if kind == "" { kind = "new" } var body any switch kind { case "new": body = protocol.NewSessionRequest{Kind: kind, Workspace: fields.Workspace, Name: optionalText(fields.Name), ProviderProfile: optionalText(fields.Provider), Model: optionalText(fields.Model), Effort: optionalText(fields.Effort)} case "fork": body = protocol.ForkSessionRequest{Kind: kind, SourceSessionID: fields.SourceSessionID, CheckpointID: fields.CheckpointID, Name: optionalText(fields.Name)} case "child": inputID, err := operationID() if err != nil { return nil, err } body = protocol.ChildSessionRequest{Kind: kind, ParentID: fields.ParentID, Address: fields.Address, Name: fields.Name, InitialInputID: inputID, Task: fields.Task, Model: optionalText(fields.Model), Effort: optionalText(fields.Effort)} default: return nil, fmt.Errorf("unknown creation kind %q", kind) } payload, err := json.Marshal(body) if err != nil { return nil, err } id, err := operationID() if err != nil { return nil, err } handle := &OperationHandle{id: id, kind: "creation", sessionID: id} handle.operation = operation{Name: "create session", BuildRequest: func(base string, body io.Reader) (*http.Request, error) { return protocol.NewCreateSessionRequestWithBody(base, id, &protocol.CreateSessionParams{IfNoneMatch: "*"}, "application/json", body) }, Body: json.RawMessage(payload), Headers: http.Header{"If-None-Match": {"*"}}, Policy: receiptRecovery, Handle: handle} return handle, nil}
type InputReceipt protocol.Input
func (receipt InputReceipt) Rejection() error { if receipt.Admission != "rejected" { return nil } err := &APIError{StatusCode: int(receipt.HTTPStatus)} if receipt.Problem != nil { err.Code, err.Message = receipt.Problem.Code, receipt.Problem.Detail if err.Message == "" { err.Message = receipt.Problem.Title } } return err}
func (receipt InputReceipt) Pending() bool { return value(receipt.Delivery) == "pending" }
func (receipt InputReceipt) Settled() bool { return value(receipt.Delivery) == "committed" || value(receipt.Delivery) == "cancelled"}
func (receipt InputReceipt) InputOrder() int64 { return value(receipt.AcceptanceOrder) }
func (receipt InputReceipt) BlockingDetail() string { if receipt.BlockingReason != nil { return receipt.BlockingReason.Detail } return ""}
func GetInput(ctx context.Context, conn *Connection, session, id string) (InputReceipt, error) { var input protocol.Input err := executeRead(ctx, conn, operation{Name: "read input", BuildRequest: func(base string, body io.Reader) (*http.Request, error) { return protocol.NewGetInputRequest(base, session, id) }, Policy: readRecovery}, func(data []byte) error { if err := decodeRequired(data, &input); err != nil { return err } if input.ID != id || input.SessionID != session { return fieldError("input identity") } return validInput(input) }) if err != nil { return InputReceipt{}, err } return InputReceipt(input), nil}func validInput(input protocol.Input) error { if input.ID == "" || input.SessionID == "" { return fieldError("input identity") } switch input.Kind { case "message", "continue", "skill", "command": default: return fieldError("input kind") } if input.Admission == "accepted" { if input.HTTPStatus != 202 || input.AcceptedAt == nil || input.AcceptanceOrder == nil || *input.AcceptanceOrder < 1 || input.Problem != nil { return fieldError("input admission") } switch value(input.Delivery) { case "pending", "committed", "cancelled": return nil } } else if input.Admission == "rejected" && input.HTTPStatus >= 400 && input.HTTPStatus <= 599 && input.Problem != nil && input.Delivery == nil && input.AcceptedAt == nil && input.AcceptanceOrder == nil && input.Turn == nil { return nil } return fieldError("input admission")}func ResolveOperation(ctx context.Context, conn *Connection, handle *OperationHandle) (InputReceipt, error) { return GetInput(ctx, conn, handle.sessionID, handle.id)}
// A found outcome is authoritative even when it rejects the input. A failed// lookup leaves admission uncertain and may permit replay of the frozen request.func executeReceipt[T any](ctx context.Context, conn *Connection, handle *OperationHandle, statuses []int, decode func([]byte, int) (T, error), resolve func(context.Context) (T, bool, error)) (T, error) { var result T if ctx == nil { ctx = context.Background() } ctx, cancel := context.WithTimeout(ctx, 20*time.Second) defer cancel() for attempt := range 2 { err := executeMutation(ctx, conn, handle.operation, statuses, func(body []byte, status int) error { var decodeErr error result, decodeErr = decode(body, status) return decodeErr }) if err == nil { return result, nil } var uncertain *UncertainOutcomeError collision := false if api, ok := errors.AsType[*APIError](err); ok { collision = handle.kind == "creation" && api.StatusCode == 412 } if !errors.As(err, &uncertain) && !collision { return result, err } if ctx.Err() != nil { return result, uncertainOperation(handle.operation, err) } resolved, found, lookupErr := resolve(ctx) if found { return resolved, lookupErr } if IsOperationExpired(lookupErr) { return result, uncertainOperation(handle.operation, lookupErr) } if handle.kind == "creation" { if api, ok := errors.AsType[*APIError](lookupErr); ok { if api.Code == "creation_conflict" { return result, lookupErr } if len(api.Decision) > 0 && api.StatusCode >= 400 && api.StatusCode < 500 && api.StatusCode != 404 { return result, lookupErr } } } if attempt == 1 { return result, uncertainOperation(handle.operation, err) } } panic("unreachable admission retry budget")}func ResolveCreation(ctx context.Context, conn *Connection, handle *OperationHandle) (Session, error) { session, err := GetSession(ctx, conn, handle.id) if err != nil { if api, ok := errors.AsType[*APIError](err); ok && len(api.Decision) > 0 { var decision protocol.CreationDecision if parseErr := decodeRequired(api.Decision, &decision); parseErr != nil { return Session{}, &ProtocolError{Code: "invalid_response", Operation: "read creation decision", Cause: parseErr} } if decision.Kind != "creation" || decision.SessionID != handle.ID() { return Session{}, fieldError("creation decision identity") } if decision.Creation != nil && !sameCreationIntent(handle.operation.Body.(json.RawMessage), decision.Creation.Submitted) { return Session{}, &APIError{StatusCode: 409, Code: "creation_conflict", Message: "the session ID belongs to another creation intent"} } } return Session{}, err } if session.wire.Creation == nil { return Session{}, fieldError("creation") } if !sameCreationIntent(handle.operation.Body.(json.RawMessage), session.wire.Creation.Submitted) { return Session{}, &APIError{StatusCode: 409, Code: "creation_conflict", Message: "the session ID belongs to another creation intent"} }
return session, nil}func IsOperationExpired(err error) bool { api, ok := errors.AsType[*APIError](err) return ok && api.StatusCode == http.StatusGone}
func sameCreationIntent(expected, stored json.RawMessage) bool { var left, right map[string]any if json.Unmarshal(expected, &left) != nil || json.Unmarshal(stored, &right) != nil { return false } var optional []string switch left["kind"] { case "new": optional = []string{"name", "provider_profile", "model", "effort"} case "fork": optional = []string{"name"} case "child": optional = []string{"model", "effort"} } for _, key := range optional { if _, ok := left[key]; !ok { left[key] = nil } } return reflect.DeepEqual(left, right)}