package 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) }